[BEAM-3754] KAFKA - Can't set commitOffsetsInFinalizeEnabled to false with KafkaIO.readBytes()

• Type: Bug
• Status: Resolved
• Priority: P3
• Resolution: Fixed
• Affects Version/s: 2.3.0
• Fix Version/s: 2.4.0
• Component/s: io-java-kafka
• Labels: patch
• Environment: Dataflow pipeline using Kafka as a Sink
• Flags: Patch
• Assignee: Raghu Angadi
• Reporter: Benjamin BENOIST (Inactive)
• Created: 27/Feb/18 16:17
• Updated: 16/May/20 13:22
• Resolved: 05/Mar/18 23:52

Description

Beam v2.3 introduces finalized offsets, in order to reduce the gaps or duplicate processing of records while restarting a pipeline.

read() sets this parameter to false by default but readBytes() doesn't, thus creating an exception:

Exception in thread "main" java.lang.IllegalStateException: Missing required properties: commitOffsetsInFinalizeEnabled
         at org.apache.beam.sdk.io.kafka.AutoValue_KafkaIO_Read$Builder.build(AutoValue_KafkaIO_Read.java:344)
         at org.apache.beam.sdk.io.kafka.KafkaIO.readBytes(KafkaIO.java:291)

The parameter can be set to true with commitOffsetsInFinalize() but never to false.

Using read() in the definition of readBytes() could prevent this kind of error in the future:

public static Read<byte[], byte[]> readBytes() {
  return read()
    .setKeyDeserializer(ByteArrayDeserializer.class)
    .setValueDeserializer(ByteArrayDeserializer.class)
    .build();
}

Issue Links

• links to: GitHub Pull Request #4792