[Bug]: KafkaIO SplittableDoFn not resuming from last committed offset

[jeanwisser]

What happened?

Using KafkaIO with ReadFromKafkaDoFn.java and commitOffsetsFinalize should commit offsets of processed messages and if the pipeline is restarted, should resume from the last committed offset.

While committing the offset works correctly, resuming from the latest committed offset does not work because the groupId used are not the same (using 2.39.0).

• Committing the offsets happens in KafkaCommitOffset.java with a consumer defined as:
  consumerFactoryFn.apply(updatedConsumerConfig)

• Reading the startOffset is done in initialRestriction() in ReadFromKafkaDoFn.java with a consumer defined as:
  consumerFactoryFn.apply( KafkaIOUtils.getOffsetConsumerConfig( "initialOffset", offsetConsumerConfig, updatedConsumerConfig)))

Since the group name of the consumer is not the same as the group name of the consumer fetching the latest offset, the pipeline will always start from the beginning of the (topic,partition) again.

Issue Priority

Priority: 2

Issue Component

Component: io-java-kafka

[chamikaramj]

@johnjcasey I believe you are looking into a similar issue ?

[johnjcasey]

I am, you can assign this to me

[chamikaramj]

Done. Feel free mark as a duplicate if needed.

[kennknowles]

Doesn't this mean that some stated functionality just isn't working? Is there a chance of data corruption if the user doesn't notice? That would make this P1 and release blocking, really.

[chamikaramj]

I think this was fixed by #22450

[johnjcasey]

It was

[Abacn]

There is still a reference to this issue in our code: Update: the reason is here:

beam/sdks/python/apache_beam/io/kafka.py

Line 191 in eb23b0a

append_args=['--experiments=use_unbounded_sdf_wrapper']))

Since it is fixed by #22450 is it safe to remove append_args=['--experiments=use_unbounded_sdf_wrapper'] below that line?