Support for dead-letter queue (DLQ) #105
Description
Use case
Currently, the sink eagerly fails and drops records records it cannot process due to row-level runtime exceptions (serialization errors, etc). As a result, it can't make progress in the source datastream until the row-level issue is fixed or worked around (such as by evolving the schema, removing the row, changing null handling mode, etc).
Describe the solution you'd like
Implement a DLQ to store unprocessed records and allow the job to progress. Expose the queue to the user in a way that makes it easy to process the impacted rows.