Hi,
I am trying to use KafkaIO with the new SplittableDoFn, but I am having trouble
understanding how offset management is supposed to work (using beam version
2.36.0).
- In the comments of ReadFromKafkaDoFn it is stated that the initial range for
KafkaSourceDescriptor startOffset can take the value of the {@code last
committed offset + 1} for the {@link Consumer#position(TopicPartition).
But when looking at the implementation of initialRestriction(), the consumer
that tries to get the committed offset is built using:
consumerFactoryFn.apply(
KafkaIOUtils.getOffsetConsumerConfig(
"initialOffset", offsetConsumerConfig, updatedConsumerConfig))
and getOffsetConsumerConfig() appends a random number to the consumer group
name. In addition to that, KafkaCommitOffset which is responsible for
committing offsets uses a different name for the consumer groupId.
How is it supposed to get the last committed offset if the group name is not
the same?
- I am also trying to commit the offsets manually at the end of my pipeline and
get KafkaIO to resume from the last committed offset by using
WatchKafkaTopicPartitionDoFn when building KafkaSourceDescriptor.of().
But since - if I understand correctly - a single (topic,partition) can be split
into multiple workers, how can we make sure that we always commit offsets in
the correct order ?
Thanks a lot for your help,
Jean.
Jean Wisser
Data Engineer
[https://www.flowtraders.com/sites/flowtraders-corp/files/images/flow-traders-logo-mv.png]
Flow Traders B.V.
T: +31 20 799 6497
F: +31 20 799 6780
Jacob Bontiusplaats 9
1018 LL Amsterdam
Nederland
www.flowtraders.com<http://www.flowtraders.com>
Flow Traders B.V. has its seat in Amsterdam, Nederland, its registered office
at Jacob Bontiusplaats 9, 1018 LL, Amsterdam, Nederland and is registered with
the Trade Registry of the Chamber of Commerce under number 33.22.3268. This
message may contain information that is not intended for you. If you are not
the addressee or if this message was sent to you by mistake, you are requested
to inform the sender and delete the message. This message may not be forwarded
or published to any other person than its addressees without Flow Traders
B.V.'s prior consent. Flow Traders B.V. accepts no liability for damage of any
kind resulting from the risks inherent in the electronic transmission of
messages.
