[Feature Request]: Enable to capture Kafka errors like JMSWriteResult

[Culby4335]

What would you like to happen?

The ability to capture the messages that fail being written to kafka because of transient errors would be nice that way we have a way of checking if the message did make it to our final destination. Right now we just assume it makes it but it would be great to capture these errors in a result to use to write to an error table that we can then check later.

The main error we are seeing is failure to connect to the Kafka ad server, we know with streaming it will auto retry but in batch we are kind of blind on what failed and what passed, but the functionality should be there for both.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[stankiewicz]

some notes,

Underlying KafkaProducer usually have some defaults on retry but eventually it will throw kafka exception
KafkaIO's KafkaWriter is capturing outcome of writing via Callback. On finish bundle, if there were exceptions, it will raise IO exception which will retry cause runner to potentially retry the bundle.
Dataflow Streaming retries forever to guarantee no data loss as there is no custom error handling in IO for network connection.
There is Bad Record handling via tag.
Dataflow Batch will cancel job after 4 failed tries.

Potential implementation:
similarly to baderrorhandler router, we could allow kafkaerrorhandler router. such handler should give full message so the user can build DLQ and retry mechanism on their own if needed.