MartijnVisser commented on code in PR #318:
URL:
https://github.com/apache/flink-connector-kafka/pull/318#discussion_r4351400857
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -976,6 +1002,241 @@ public void testCheckSourceIntegrityFromProperties()
throws Exception {
}
}
+ /**
+ * A bounded enumerator that is restored with every subscribed partition
already assigned must
+ * still tell its readers that no more splits are coming. The restored
enumerator runs its
+ * one-time discovery, finds nothing new, and on an unfixed enumerator
returns early before
+ * reaching the only place that sets {@code noMoreNewPartitionSplits}, so
the readers wait for a
+ * {@code NoMoreSplitsEvent} that never arrives and the job never finishes.
+ *
+ * <p>Do not weaken this test by enabling partition discovery or by
leaving a partition out of
+ * the restored state: either makes the partition change non-empty and the
early return is no
+ * longer taken.
+ */
+ @Test
+ public void
testRestoredBoundedEnumeratorSignalsNoMoreSplitsWithoutPartitionChanges()
Review Comment:
Agreed, they only differ in the enumerator state. It's one parameterised
test now, with a named case for each.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -976,6 +1002,241 @@ public void testCheckSourceIntegrityFromProperties()
throws Exception {
}
}
+ /**
+ * A bounded enumerator that is restored with every subscribed partition
already assigned must
+ * still tell its readers that no more splits are coming. The restored
enumerator runs its
+ * one-time discovery, finds nothing new, and on an unfixed enumerator
returns early before
+ * reaching the only place that sets {@code noMoreNewPartitionSplits}, so
the readers wait for a
+ * {@code NoMoreSplitsEvent} that never arrives and the job never finishes.
+ *
+ * <p>Do not weaken this test by enabling partition discovery or by
leaving a partition out of
+ * the restored state: either makes the partition change non-empty and the
early return is no
+ * longer taken.
+ */
+ @Test
+ public void
testRestoredBoundedEnumeratorSignalsNoMoreSplitsWithoutPartitionChanges()
+ throws Throwable {
+ Set<KafkaPartitionSplit> restoredAssignedSplits =
+
KafkaSourceTestEnv.getPartitionsForTopics(PRE_EXISTING_TOPICS).stream()
+ .map(tp -> new KafkaPartitionSplit(tp, 0L))
+ .collect(Collectors.toSet());
+ assertThat(restoredAssignedSplits).isNotEmpty();
+
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ -1,
+ PRE_EXISTING_TOPICS,
+ restoredAssignedSplits,
+ Collections.emptySet(),
+ new Properties(),
+ OffsetsInitializer.earliest(),
+ Boundedness.BOUNDED,
+ true)) {
+ enumerator.start();
+
+ // A reader that re-registers before the discovery callback
returns.
+ registerReader(context, enumerator, READER0);
+ runOneTimePartitionDiscovery(context);
+ // A reader that re-registers after it.
+ registerReader(context, enumerator, READER1);
+
+ assertThat(context.getSplitsAssignmentSequence())
+ .as("Every partition is already assigned, so nothing may
be assigned again")
+ .isEmpty();
+ SoftAssertions.assertSoftly(
+ softly -> {
+ softly.assertThat(context.hasNoMoreSplits(READER0))
Review Comment:
Good point. The test now records the events and checks that each reader is
signalled exactly once.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -976,6 +1002,241 @@ public void testCheckSourceIntegrityFromProperties()
throws Exception {
}
}
+ /**
+ * A bounded enumerator that is restored with every subscribed partition
already assigned must
+ * still tell its readers that no more splits are coming. The restored
enumerator runs its
+ * one-time discovery, finds nothing new, and on an unfixed enumerator
returns early before
+ * reaching the only place that sets {@code noMoreNewPartitionSplits}, so
the readers wait for a
+ * {@code NoMoreSplitsEvent} that never arrives and the job never finishes.
+ *
+ * <p>Do not weaken this test by enabling partition discovery or by
leaving a partition out of
+ * the restored state: either makes the partition change non-empty and the
early return is no
Review Comment:
You're right, with discovery enabled `start()` schedules a periodic callable
and the test hangs waiting for the one-time one. That half was wrong and is
gone.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -976,6 +1002,241 @@ public void testCheckSourceIntegrityFromProperties()
throws Exception {
}
}
+ /**
+ * A bounded enumerator that is restored with every subscribed partition
already assigned must
+ * still tell its readers that no more splits are coming. The restored
enumerator runs its
+ * one-time discovery, finds nothing new, and on an unfixed enumerator
returns early before
+ * reaching the only place that sets {@code noMoreNewPartitionSplits}, so
the readers wait for a
+ * {@code NoMoreSplitsEvent} that never arrives and the job never finishes.
+ *
+ * <p>Do not weaken this test by enabling partition discovery or by
leaving a partition out of
+ * the restored state: either makes the partition change non-empty and the
early return is no
+ * longer taken.
+ */
+ @Test
+ public void
testRestoredBoundedEnumeratorSignalsNoMoreSplitsWithoutPartitionChanges()
+ throws Throwable {
+ Set<KafkaPartitionSplit> restoredAssignedSplits =
+
KafkaSourceTestEnv.getPartitionsForTopics(PRE_EXISTING_TOPICS).stream()
+ .map(tp -> new KafkaPartitionSplit(tp, 0L))
+ .collect(Collectors.toSet());
+ assertThat(restoredAssignedSplits).isNotEmpty();
+
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ -1,
+ PRE_EXISTING_TOPICS,
+ restoredAssignedSplits,
+ Collections.emptySet(),
+ new Properties(),
+ OffsetsInitializer.earliest(),
+ Boundedness.BOUNDED,
+ true)) {
+ enumerator.start();
+
+ // A reader that re-registers before the discovery callback
returns.
+ registerReader(context, enumerator, READER0);
+ runOneTimePartitionDiscovery(context);
+ // A reader that re-registers after it.
+ registerReader(context, enumerator, READER1);
+
+ assertThat(context.getSplitsAssignmentSequence())
+ .as("Every partition is already assigned, so nothing may
be assigned again")
+ .isEmpty();
+ SoftAssertions.assertSoftly(
+ softly -> {
+ softly.assertThat(context.hasNoMoreSplits(READER0))
+ .as("Reader registered before the discovery
callback")
+ .isTrue();
+ softly.assertThat(context.hasNoMoreSplits(READER1))
+ .as("Reader registered after the discovery
callback")
+ .isTrue();
+ });
+ }
+ }
+
+ /**
+ * The mirror of the two tests above. An unbounded source with one-time
discovery cannot act on
+ * an empty partition change: it has no readers to tell that the input
ended, and treating the
Review Comment:
Badly put, I meant that an unbounded source has nothing to tell its readers.
Rewritten.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -976,6 +1002,241 @@ public void testCheckSourceIntegrityFromProperties()
throws Exception {
}
}
+ /**
+ * A bounded enumerator that is restored with every subscribed partition
already assigned must
+ * still tell its readers that no more splits are coming. The restored
enumerator runs its
+ * one-time discovery, finds nothing new, and on an unfixed enumerator
returns early before
+ * reaching the only place that sets {@code noMoreNewPartitionSplits}, so
the readers wait for a
+ * {@code NoMoreSplitsEvent} that never arrives and the job never finishes.
+ *
+ * <p>Do not weaken this test by enabling partition discovery or by
leaving a partition out of
+ * the restored state: either makes the partition change non-empty and the
early return is no
+ * longer taken.
+ */
+ @Test
+ public void
testRestoredBoundedEnumeratorSignalsNoMoreSplitsWithoutPartitionChanges()
+ throws Throwable {
+ Set<KafkaPartitionSplit> restoredAssignedSplits =
+
KafkaSourceTestEnv.getPartitionsForTopics(PRE_EXISTING_TOPICS).stream()
+ .map(tp -> new KafkaPartitionSplit(tp, 0L))
+ .collect(Collectors.toSet());
+ assertThat(restoredAssignedSplits).isNotEmpty();
+
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ -1,
+ PRE_EXISTING_TOPICS,
+ restoredAssignedSplits,
+ Collections.emptySet(),
+ new Properties(),
+ OffsetsInitializer.earliest(),
+ Boundedness.BOUNDED,
+ true)) {
+ enumerator.start();
+
+ // A reader that re-registers before the discovery callback
returns.
+ registerReader(context, enumerator, READER0);
+ runOneTimePartitionDiscovery(context);
+ // A reader that re-registers after it.
+ registerReader(context, enumerator, READER1);
+
+ assertThat(context.getSplitsAssignmentSequence())
+ .as("Every partition is already assigned, so nothing may
be assigned again")
+ .isEmpty();
+ SoftAssertions.assertSoftly(
+ softly -> {
+ softly.assertThat(context.hasNoMoreSplits(READER0))
+ .as("Reader registered before the discovery
callback")
+ .isTrue();
+ softly.assertThat(context.hasNoMoreSplits(READER1))
+ .as("Reader registered after the discovery
callback")
+ .isTrue();
+ });
+ }
+ }
+
+ /**
+ * The mirror of the two tests above. An unbounded source with one-time
discovery cannot act on
Review Comment:
It wasn't, one of them is below. The Javadoc now links the bounded test by
name.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -976,6 +1002,241 @@ public void testCheckSourceIntegrityFromProperties()
throws Exception {
}
}
+ /**
+ * A bounded enumerator that is restored with every subscribed partition
already assigned must
+ * still tell its readers that no more splits are coming. The restored
enumerator runs its
+ * one-time discovery, finds nothing new, and on an unfixed enumerator
returns early before
+ * reaching the only place that sets {@code noMoreNewPartitionSplits}, so
the readers wait for a
+ * {@code NoMoreSplitsEvent} that never arrives and the job never finishes.
+ *
+ * <p>Do not weaken this test by enabling partition discovery or by
leaving a partition out of
+ * the restored state: either makes the partition change non-empty and the
early return is no
+ * longer taken.
+ */
+ @Test
+ public void
testRestoredBoundedEnumeratorSignalsNoMoreSplitsWithoutPartitionChanges()
+ throws Throwable {
+ Set<KafkaPartitionSplit> restoredAssignedSplits =
+
KafkaSourceTestEnv.getPartitionsForTopics(PRE_EXISTING_TOPICS).stream()
+ .map(tp -> new KafkaPartitionSplit(tp, 0L))
+ .collect(Collectors.toSet());
+ assertThat(restoredAssignedSplits).isNotEmpty();
+
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ -1,
+ PRE_EXISTING_TOPICS,
+ restoredAssignedSplits,
+ Collections.emptySet(),
+ new Properties(),
+ OffsetsInitializer.earliest(),
+ Boundedness.BOUNDED,
+ true)) {
+ enumerator.start();
+
+ // A reader that re-registers before the discovery callback
returns.
+ registerReader(context, enumerator, READER0);
+ runOneTimePartitionDiscovery(context);
+ // A reader that re-registers after it.
+ registerReader(context, enumerator, READER1);
+
+ assertThat(context.getSplitsAssignmentSequence())
+ .as("Every partition is already assigned, so nothing may
be assigned again")
+ .isEmpty();
+ SoftAssertions.assertSoftly(
+ softly -> {
+ softly.assertThat(context.hasNoMoreSplits(READER0))
+ .as("Reader registered before the discovery
callback")
+ .isTrue();
+ softly.assertThat(context.hasNoMoreSplits(READER1))
+ .as("Reader registered after the discovery
callback")
+ .isTrue();
+ });
+ }
+ }
+
+ /**
+ * The mirror of the two tests above. An unbounded source with one-time
discovery cannot act on
+ * an empty partition change: it has no readers to tell that the input
ended, and treating the
+ * discovery as finished would make a later restore initialise the
partitions that appeared
+ * meanwhile from the earliest offset instead of the configured one. It
must keep returning
+ * early, and this test fails if the condition in checkPartitionChanges is
ever widened past
+ * bounded sources.
+ */
+ @Test
+ public void testUnboundedSourceKeepsSkippingAnEmptyPartitionChange()
throws Throwable {
+ Collection<String> noSuchTopic =
Collections.singleton("topic-that-does-not-exist");
+
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ -1,
+ noSuchTopic,
+ Collections.emptySet(),
+ Collections.emptySet(),
+ new Properties(),
+ OffsetsInitializer.earliest(),
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ false)) {
+ enumerator.start();
+ registerReader(context, enumerator, READER0);
+ runOneTimePartitionDiscovery(context);
+
+ SoftAssertions.assertSoftly(
+ softly -> {
+ try {
+ softly.assertThat(
+
enumerator.snapshotState(1L).initialDiscoveryFinished())
+ .as("An empty change must not mark the
discovery as finished")
+ .isFalse();
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ softly.assertThat(context.hasNoMoreSplits(READER0))
Review Comment:
Right, `signalNoMoreSplits` is only reached for bounded sources, so that
assertion couldn't fail. The test now only checks `initialDiscoveryFinished`,
which is what breaks if the condition is widened.
--
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]
