You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/10/11 07:08:01 UTC
[flink-connector-kafka] branch v3.0 updated: [FLINK-33231] [source] Properly evict offsetsToCommit cache on checkpoint complete if no offsets exist
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch v3.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/v3.0 by this push:
new b316d8ce [FLINK-33231] [source] Properly evict offsetsToCommit cache on checkpoint complete if no offsets exist
b316d8ce is described below
commit b316d8ce25cc63bc6cdb858831e8566164a16b81
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Tue Oct 10 12:18:19 2023 -0700
[FLINK-33231] [source] Properly evict offsetsToCommit cache on checkpoint complete if no offsets exist
Prior to this fix, if the offsets to commit for a given checkpoint is empty,
which can be the case if no starting offsets were retrieved from Kafka yet,
then on checkpoint completion the cache is not properly evicted up to the
given checkpoint.
This change fixes this such that in notifyOnCheckpointComplete, we shortcut
the method execution to not need to try to commit the offsets since its
empty anyways, and always remember to evict the cache up to the completed
checkpoint.
(cherry picked from commit b0f15f279915a3862d353c9d2726a52741c248ba)
---
.../kafka/source/reader/KafkaSourceReader.java | 21 ++++++++++++++-------
.../kafka/source/reader/KafkaSourceReaderTest.java | 22 +++++++++++++++++++++-
2 files changed, 35 insertions(+), 8 deletions(-)
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
index e8fe3d77..e86cba00 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
@@ -133,9 +133,13 @@ public class KafkaSourceReader<T>
Map<TopicPartition, OffsetAndMetadata> committedPartitions =
offsetsToCommit.get(checkpointId);
if (committedPartitions == null) {
- LOG.debug(
- "Offsets for checkpoint {} either do not exist or have already been committed.",
- checkpointId);
+ LOG.debug("Offsets for checkpoint {} have already been committed.", checkpointId);
+ return;
+ }
+
+ if (committedPartitions.isEmpty()) {
+ LOG.debug("There are no offsets to commit for checkpoint {}.", checkpointId);
+ removeAllOffsetsToCommitUpToCheckpoint(checkpointId);
return;
}
@@ -168,14 +172,17 @@ public class KafkaSourceReader<T>
entry ->
committedPartitions.containsKey(
entry.getKey()));
- while (!offsetsToCommit.isEmpty()
- && offsetsToCommit.firstKey() <= checkpointId) {
- offsetsToCommit.remove(offsetsToCommit.firstKey());
- }
+ removeAllOffsetsToCommitUpToCheckpoint(checkpointId);
}
});
}
+ private void removeAllOffsetsToCommitUpToCheckpoint(long checkpointId) {
+ while (!offsetsToCommit.isEmpty() && offsetsToCommit.firstKey() <= checkpointId) {
+ offsetsToCommit.remove(offsetsToCommit.firstKey());
+ }
+ }
+
@Override
protected KafkaPartitionSplitState initializedState(KafkaPartitionSplit split) {
return new KafkaPartitionSplitState(split);
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
index e4ee39d2..9aa478ee 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
@@ -59,6 +59,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -183,7 +184,26 @@ public class KafkaSourceReaderTest extends SourceReaderTestBase<KafkaPartitionSp
(KafkaSourceReader<Integer>)
createReader(Boundedness.CONTINUOUS_UNBOUNDED, groupId)) {
reader.snapshotState(100L);
- reader.notifyCheckpointComplete(100L);
+ reader.snapshotState(101L);
+ reader.snapshotState(102L);
+
+ // After each snapshot, a new entry should have been added to the offsets-to-commit
+ // cache for the checkpoint
+ final Map<Long, Map<TopicPartition, OffsetAndMetadata>> expectedOffsetsToCommit =
+ new HashMap<>();
+ expectedOffsetsToCommit.put(100L, new HashMap<>());
+ expectedOffsetsToCommit.put(101L, new HashMap<>());
+ expectedOffsetsToCommit.put(102L, new HashMap<>());
+ assertThat(reader.getOffsetsToCommit()).isEqualTo(expectedOffsetsToCommit);
+
+ // only notify up to checkpoint 101L; all offsets prior to 101L should be evicted from
+ // cache, leaving only 102L
+ reader.notifyCheckpointComplete(101L);
+
+ final Map<Long, Map<TopicPartition, OffsetAndMetadata>>
+ expectedOffsetsToCommitAfterNotify = new HashMap<>();
+ expectedOffsetsToCommitAfterNotify.put(102L, new HashMap<>());
+ assertThat(reader.getOffsetsToCommit()).isEqualTo(expectedOffsetsToCommitAfterNotify);
}
// Verify the committed offsets.
try (AdminClient adminClient = KafkaSourceTestEnv.getAdminClient()) {