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 2026/09/10 12:23:34 UTC
(flink-connector-kafka) branch main updated: [FLINK-40362][connector/kafka] Fix idle reader hang when no-more-splits precedes metadata update (#291)
This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 48ddbf1f [FLINK-40362][connector/kafka] Fix idle reader hang when no-more-splits precedes metadata update (#291)
48ddbf1f is described below
commit 48ddbf1fd5d6d3d60ed30ec4ab0780f70f9cbc51
Author: Sylwester Lachiewicz <sl...@apache.org>
AuthorDate: Thu Sep 10 14:23:28 2026 +0200
[FLINK-40362][connector/kafka] Fix idle reader hang when no-more-splits precedes metadata update (#291)
An idle reader (subtask with no assigned splits) that receives the no-more-splits
signal before the reply to GetMetadataUpdateEvent never forwards it to the sub-readers
created by that update: the re-notification in handleSourceEvents was guarded by
!pendingSplits.isEmpty(), which is false in exactly this case. The reader then
returns NOTHING_AVAILABLE indefinitely and a bounded job never finishes.
Re-deliver the no-more-splits signal after processing the reader's first metadata update
(captured via !isActivelyConsumingSplits before the flag flips). Replaying only on the
first metadata update ensures that an active reader waiting for splits on newly added
clusters/topics does not terminate prematurely before those splits and the subsequent
no-more-splits signal arrive. notifyNoMoreSplits() is idempotent and only fires when
isNoMoreSplits is true.
Verified:
- DynamicKafkaSourceReaderTest#testIdleReaderFinishesWhenNoMoreSplitsArrivesBeforeMetadata:
fails on main (NOTHING_AVAILABLE), passes with this change (END_OF_INPUT).
- DynamicKafkaSourceReaderTest#testActiveReaderWaitsForNewSplitsAfterMetadataChange:
passes on main, fails if re-delivery is unconditional on every metadata update,
passes with the firstMetadataUpdate guard.
- SourceTestSuiteBase#testIdleReader through DynamicKafkaSourceITTest$IntegrationTests
hangs on unfixed main whenever a run leaves a subtask idle.
Generated-by: Claude Fable 5
---
.../source/reader/DynamicKafkaSourceReader.java | 16 ++++--
.../reader/DynamicKafkaSourceReaderTest.java | 61 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 3 deletions(-)
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
index 2a6bd266..0d2c9c73 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
@@ -378,6 +378,10 @@ public class DynamicKafkaSourceReader<T> implements SourceReader<T, DynamicKafka
clustersProperties.putAll(newClustersProperties);
}
+ // Captured before the flag flips below, so the no-more-splits replay can tell the
+ // reader's first metadata update from a later metadata change.
+ final boolean firstMetadataUpdate = !isActivelyConsumingSplits;
+
// finally mark the reader as active, if not already and add pending splits
if (!isActivelyConsumingSplits) {
isActivelyConsumingSplits = true;
@@ -399,9 +403,15 @@ public class DynamicKafkaSourceReader<T> implements SourceReader<T, DynamicKafka
addSplits(validPendingSplits);
pendingSplits.clear();
- if (isNoMoreSplits) {
- notifyNoMoreSplits();
- }
+ }
+
+ // Replay only on the first metadata update. On a later metadata change the reader must
+ // wait for the enumerator to signal again after the new assignments (that re-signal is
+ // currently missing for a sub-enumerator recreated with partitions already assigned; see
+ // FLINK-31006), so replaying here would finish an active reader before the new topic's
+ // splits arrive.
+ if (isNoMoreSplits && firstMetadataUpdate) {
+ notifyNoMoreSplits();
}
// Releasing the last split output can expose the runtime output as idle immediately. Keep
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
index ebffa3e9..1ef6f23b 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
@@ -370,6 +370,67 @@ public class DynamicKafkaSourceReaderTest extends SourceReaderTestBase<DynamicKa
}
}
+ @Test
+ void testIdleReaderFinishesWhenNoMoreSplitsArrivesBeforeMetadata() throws Exception {
+ TestingReaderContext context = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<Integer> reader = createReaderWithoutStart(context)) {
+ TrackingReaderOutput<Integer> readerOutput = new TrackingReaderOutput<>();
+ reader.start();
+
+ // The enumerator signals no more splits before the reader receives the metadata
+ // update event, e.g. when an idle reader registers after all bounded
+ // sub-enumerators have already finished split discovery.
+ reader.notifyNoMoreSplits();
+
+ MetadataUpdateEvent metadata =
+ DynamicKafkaSourceTestHelper.getMetadataUpdateEvent(TOPIC);
+ CompletableFuture<Void> availableBeforeMetadata = reader.isAvailable();
+ reader.handleSourceEvents(metadata);
+
+ assertThat(availableBeforeMetadata)
+ .as("the metadata update must wake up a task parked on the earlier future")
+ .isDone();
+ assertThat(reader.pollNext(readerOutput))
+ .as(
+ "idle reader must reach END_OF_INPUT even when no-more-splits precedes the metadata update event")
+ .isEqualTo(InputStatus.END_OF_INPUT);
+ }
+ }
+
+ @Test
+ void testActiveReaderWaitsForNewSplitsAfterMetadataChange() throws Exception {
+ TestingReaderContext context = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<Integer> reader = createReaderWithoutStart(context)) {
+ TrackingReaderOutput<Integer> readerOutput = new TrackingReaderOutput<>();
+ reader.start();
+
+ // First metadata update: only cluster 0 is known, so the reader goes active with a
+ // single sub-reader.
+ KafkaStream clusterZeroOnly = DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ clusterZeroOnly.getClusterMetadataMap().remove(kafkaClusterId1);
+ reader.handleSourceEvents(
+ new MetadataUpdateEvent(Collections.singleton(clusterZeroOnly)));
+
+ // The enumerator finished discovery for that metadata epoch.
+ reader.notifyNoMoreSplits();
+
+ // Cluster 1 appears. The reader holds no splits, so the metadata change recreates every
+ // sub-reader and only the reader-level flag still remembers the earlier signal. Splits
+ // and a fresh no-more-splits signal for the new metadata arrive after this event.
+ reader.handleSourceEvents(DynamicKafkaSourceTestHelper.getMetadataUpdateEvent(TOPIC));
+
+ assertThat(reader.pollNext(readerOutput))
+ .as(
+ "reader must not finish on the stale no-more-splits signal while a newly added cluster still awaits its splits")
+ .isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
+ reader.notifyNoMoreSplits();
+ assertThat(reader.pollNext(readerOutput))
+ .as("reader finishes once the enumerator signals again for the new metadata")
+ .isEqualTo(InputStatus.END_OF_INPUT);
+ }
+ }
+
@Test
void testAvailabilityFutureUpdates() throws Exception {
TestingReaderContext context = new TestingReaderContext();