dalelane commented on code in PR #293:
URL:
https://github.com/apache/flink-connector-kafka/pull/293#discussion_r3842335257
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -125,6 +131,7 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>>
fetch() throws IOExce
for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
long consumerPosition = getConsumerPosition(tp, "retrieving
consumer position");
+ lastKnownPositions.put(tp, consumerPosition);
Review Comment:
@AHeise - Thanks for the review.
I went with the split reader to keep track of the consumer's position as it
polls, so I could correct offsets as they're committed to Kafka. By leaving it
to the last minute like this, when the checkpoint completes the split reader
can take the offset the reader has reached, and correct it using the most
recent position the consumer reported.
I think what you're describing would identify the position earlier, as
records are emitted. By the time the checkpoint starts, the value would be
fixed, so the commit will send what has already been decided.
The difference comes down to when we get the offset right. Deciding late
means the committed offset will always reflect the most recent poll and will
always be right at every checkpoint. Deciding early means it could be one poll
out of date, as corrections will land once the reader has read past the records
it already has, so it needs another record to arrive to carry an updated
position through.
For busy topics, no-one will notice the difference. But what prompted me to
pick up this issue was a problem report on a low-traffic / bursty transactional
topic, where the offset effectively reported an incorrect lag "forever". (In
hindsight, I should've written a test for this scenario, then I could point at
it now! That's my bad, and even if we keep my design I should add a test for
that.)
The question is whether solving that issue is worth the cost/complexity of
my approach. IMHO it is, but I'm open to debate :-)
(_@AHeise - I'm aware I'm making a bunch of assumptions from your
description, so it's entirely possible I've misunderstood what you're
suggesting. Would a quick agent-driven spike to give it a try and see what an
implementation would look like help?_)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
