dalelane commented on code in PR #333:
URL:
https://github.com/apache/flink-connector-kafka/pull/333#discussion_r4514144740
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java:
##########
@@ -98,6 +98,31 @@ public void commitOffsets(
}
}
+ /**
+ * Commits the given offsets synchronously when the fetcher closes its
consumer, after {@link
+ * #close(long)} shut the fetcher down. Must be called before {@link
#close(long)}.
+ */
+ public void commitOffsetsOnClose(
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
+ Set<TopicPartition> partitionsWithoutOffset) {
+ // A fetcher shut down by close() or by an error stays listed until it
exits, so try each
+ // of them.
+ for (SplitFetcher<ConsumerRecord<byte[], byte[]>, KafkaPartitionSplit>
fetcher :
+ fetchers.values()) {
+ if (((KafkaPartitionSplitReader) fetcher.getSplitReader())
+ .commitOffsetsOnClose(offsetsToCommit,
partitionsWithoutOffset)) {
+ return;
+ }
+ }
+ // No fetcher runs, or all are already closing their consumers. Only a
fetcher thread may
+ // use a consumer, so start one that commits as it closes.
+ final SplitFetcher<ConsumerRecord<byte[], byte[]>,
KafkaPartitionSplit> splitFetcher =
+ createSplitFetcher();
Review Comment:
In my earlier comment I said the close commit runs in the same
KafkaPartitionSplitReader that fetched the records, so reconciliation has the
state it needs. That's true when a fetcher is still running, but I missed this
path.
When no fetcher is left, for example after a fetcher fails and the job fails
over, this starts a new one. Its split reader has no memory of earlier commits,
so it can't stop the committed offset from moving backwards:
1. A checkpoint holds offset 1. Reconciliation moves it past the commit
marker, and 2 is committed.
2. The next checkpoint also holds 1, and completes. Its commit is queued.
3. The fetcher fails before running that commit.
4. On close, a new fetcher commits 1. The committed offset goes back from 2
to 1.
The gap isn't new: commitOffsets already starts fresh fetchers like this,
and my PR only stopped the offset moving backwards while a partition stays
assigned to a reader. But this commit adds another way in.
Could a fetcher started only to commit check the group's committed offsets
first (consumer.committed), and skip anything lower?
Checking first would also mean that in cases where there's nothing it can
commit (partitions without an offset) it'd avoid unnecessarily creating a
consumer
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -64,6 +66,10 @@
public class KafkaPartitionSplitReader
implements SplitReader<ConsumerRecord<byte[], byte[]>,
KafkaPartitionSplit> {
private static final Logger LOG =
LoggerFactory.getLogger(KafkaPartitionSplitReader.class);
+ // The commit returns once Kafka acknowledges it, and only an unreachable
group coordinator
+ // makes it delay closing the consumer by this long. The reader waits for
its fetcher to close
+ // for at most source.reader.close.timeout, 30 seconds by default.
+ private static final Duration COMMIT_ON_CLOSE_TIMEOUT =
Duration.ofSeconds(10);
Review Comment:
I agree that this would [usually be within
milliseconds](https://github.com/apache/flink-connector-kafka/pull/333/changes#r4486512435),
given that it returns as soon as the broker acks. But what happens when there
is no ack from the group coordinator - if the group coordinator can't be
reached or has stopped responding while the job is stopping or falling over?
(With the classic protocol, that is when the latest async commit is still in
flight)
In that case, closing the fetcher does this:
(10 seconds after commit request) `commitSync` gives up - the request is
still in flight, `consumer.close()` starts waiting for it
(30 seconds after commit request) commit request times out
(request.timeout.ms). `close()` stops waiting and returns
This means the fetcher can take about 30 seconds to close. The reader only
waits 30 seconds (source.reader.close.timeout) then logs and continues. The
fetcher thread keeps running in the background with the consumer.
The comment on COMMIT_ON_CLOSE_TIMEOUT suggest the aim was for the commit to
fit in the time budget, but this doesn't count `consumer.close()`
This matters beyond stop-with-savepoint, because the commit runs on every
close (failover, cancel, and DynamicKafkaSourceReader restarting its
sub-readers). Failovers caused by broker trouble could hit this.
DynamicKafkaSourceReader closes its sub-readers one at a time on the task
thread, so each cluster that has stopped responding adds its own wait.
Could the commit and consumer.close() share what's left of the close budget,
so together they stay inside source.reader.close.timeout?
(And it'd help to make it clear in the PR description that this applies to
every close, not only stop-with-savepoint)
--
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]
