[BEAM-7870] Externally configured KafkaIO / PubsubIO consumer causes coder problems

• Type: Bug
• Status: Open
• Priority: P3
• Resolution: Unresolved
• Component/s: runner-flink, sdk-java-core
• Labels: Clarified
• Assignee: Unassigned
• Reporter: Maximilian Michels
• Created: 01/Aug/19 09:11
• Updated: 04/Jun/22 12:57

Description

There are limitations for the consumer to work correctly. The biggest issue is the structure of KafkaIO itself, which uses a combination of the source interface and DoFns to generate the desired output. The problem is that the source interface is natively translated by the Flink Runner to support unbounded sources in portability, while the DoFn runs in a Java environment.

To transfer data between the two a coder needs to be involved. It happens to be that the initial read does not immediately drop the KafakRecord structure which does not work together well with our current assumption of only supporting "standard coders" present in all SDKs. Only the subsequent DoFn converts the KafkaRecord structure into a raw KV[byte, byte], but the DoFn won't have the coder available in its environment.

There are several possible solutions:
1. Make the DoFn which drops the KafkaRecordCoder a native Java transform in the Flink Runner
2. Modify KafkaIO to immediately drop the KafkaRecord structure
3. Add the KafkaRecordCoder to all SDKs
4. Add a generic coder, e.g. AvroCoder to all SDKs

For a workaround which uses (3), please see this patch which is not a proper fix but adds KafkaRecordCoder to the SDK such that it can be used encode/decode records: https://github.com/mxm/beam/commit/b31cf99c75b3972018180d8ccc7e73d311f4cfed

 

See also https://github.com/apache/beam/pull/8251

Issue Links

• is duplicated by: BEAM-9046 Kafka connector for Python throws ClassCastException when reading KafkaRecord