Savonitar commented on code in PR #334:
URL:
https://github.com/apache/flink-connector-kafka/pull/334#discussion_r4426849407
##########
docs/content/docs/connectors/table/upsert-kafka.md:
##########
@@ -195,6 +195,22 @@ Connector Options
format which means that key columns appear in the data type for both
the key and value format.
</td>
</tr>
+ <tr>
+ <td><h5>scan.topic-partition-discovery.interval</h5></td>
+ <td>optional</td>
+ <td>no</td>
Review Comment:
this cell lands in the incorrect column if we check rendered version
https://github.com/Efrat19/flink-connector-kafka/blob/3b1edb12b23e0af5af01dee09f909d29e7b2ce93/docs/content/docs/connectors/table/upsert-kafka.md#connector-options:
<img width="929" height="529" alt="Image"
src="https://github.com/user-attachments/assets/3fccf495-a02c-4aeb-a643-c137e57622ae";
/>
The upsert-kafka table has five columns and there is no "Forwarded" column,
but both new rows have six cells.
##########
docs/content.zh/docs/connectors/table/upsert-kafka.md:
##########
@@ -183,6 +183,22 @@ of all available metadata fields.
</ul>
</td>
</tr>
+ <tr>
+ <td><h5>scan.topic-partition-discovery.interval</h5></td>
+ <td>optional</td>
+ <td>no</td>
Review Comment:
same
https://github.com/apache/flink-connector-kafka/pull/334/changes#r4426849407
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactory.java:
##########
@@ -166,6 +168,10 @@ public DynamicTableSource createDynamicTableSource(Context
context) {
KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
Long.toString(partitionDiscoveryInterval.toMillis()));
+ properties.setProperty(
Review Comment:
Same
https://github.com/apache/flink-connector-kafka/pull/334/changes#r4427595622
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java:
##########
@@ -848,6 +851,197 @@ public void
testEnumeratorStateDoesNotContainStaleTopicPartitions() throws Throw
}
}
+ @Test
+ public void testTrackedTopicIdsSurviveMetadataChangeAndDetectRecreation()
throws Throwable {
+ final String topic2 = TOPIC + "_integrity";
+
+ DynamicKafkaSourceTestHelper.createTopic(topic2,
NUM_SPLITS_PER_CLUSTER, 1);
+ DynamicKafkaSourceTestHelper.produceToKafka(
+ topic2, NUM_SPLITS_PER_CLUSTER, NUM_RECORDS_PER_SPLIT);
+ final String clusterId =
DynamicKafkaSourceTestHelper.getKafkaClusterId(0);
+
+ final Set<KafkaStream> initialStreams =
+ Collections.singleton(
+ new KafkaStream(
+ TOPIC,
+
DynamicKafkaSourceTestHelper.getClusterMetadataMap(
+ 0, TOPIC, topic2)));
+ final Set<KafkaStream> updatedStreams =
+ Collections.singleton(
+ new KafkaStream(
+ TOPIC,
+
DynamicKafkaSourceTestHelper.getClusterMetadataMap(0, TOPIC)));
+
+ final DynamicKafkaSourceEnumState stateAfterMetadataChange;
+ final Map<String, String> initialTrackedIds;
+ try (MockKafkaMetadataService metadataService =
+ new MockKafkaMetadataService(initialStreams);
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ metadataService,
+
DynamicKafkaSourceEnumeratorTest::enableTopicIntegrityCheck)) {
+ enumerator.start();
+ context.runPeriodicCallable(0);
+ runAllOneTimeCallables(context);
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
0);
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
1);
+
+ initialTrackedIds =
getTrackedTopicIds(enumerator.snapshotState(-1));
+ assertThat(initialTrackedIds).containsOnlyKeys(TOPIC, topic2);
+
+ metadataService.setKafkaStreams(updatedStreams);
+ context.runPeriodicCallable(0);
+
+ // the restarted enumerators have not rediscovered the topics yet,
so the ids can only
+ // come from the state carried over the metadata change
+ assertThat(getTrackedTopicIds(enumerator.snapshotState(-1)))
+ .containsOnly(Map.entry(TOPIC,
initialTrackedIds.get(TOPIC)));
+
+ // the rediscovery agrees with the carried over id, so the check
passes
+ runAllOneTimeCallables(context);
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
2);
+
+ stateAfterMetadataChange = enumerator.snapshotState(-1);
+ assertThat(getTrackedTopicIds(stateAfterMetadataChange))
+ .containsOnly(Map.entry(TOPIC,
initialTrackedIds.get(TOPIC)));
+ }
+
+ // a stale id in the checkpoint is what a topic recreated after the
checkpoint looks like
+ final KafkaSourceEnumState clusterState =
+
stateAfterMetadataChange.getClusterEnumeratorStates().get(clusterId);
+ final String staleTopicId = Uuid.randomUuid().toString();
+ final DynamicKafkaSourceEnumState staleState =
+ withClusterState(
+ stateAfterMetadataChange,
+ clusterId,
+ new KafkaSourceEnumState(
+ clusterState.splits(),
+ clusterState.initialDiscoveryFinished(),
+ Collections.singletonMap(TOPIC,
staleTopicId)));
+
+ // topic2 is back in the stream, so the restored enumerators are
restarted on the first
+ // metadata refresh before they run their first discovery
+ try (MockKafkaMetadataService metadataService =
+ new MockKafkaMetadataService(initialStreams);
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ metadataService,
+
DynamicKafkaSourceEnumeratorTest::enableTopicIntegrityCheck,
+ staleState)) {
+ enumerator.start();
+ context.runPeriodicCallable(0);
+
+ assertThat(getTrackedTopicIds(enumerator.snapshotState(-1)))
+ .containsOnly(Map.entry(TOPIC, staleTopicId));
+ assertThatThrownBy(() -> runAllOneTimeCallables(context))
+ .hasRootCauseInstanceOf(TopicIntegrityException.class)
+ .hasRootCauseMessage("Topic " + TOPIC + " was recreated");
+ }
+ }
+
+ @Test
+ public void testRestoreKeepsTrackedTopicIdsOfTopicsWithUnassignedSplits()
throws Throwable {
+ final String topic2 = TOPIC + "_unassigned";
+ DynamicKafkaSourceTestHelper.createTopic(topic2,
NUM_SPLITS_PER_CLUSTER, 1);
+ DynamicKafkaSourceTestHelper.produceToKafka(
+ topic2, NUM_SPLITS_PER_CLUSTER, NUM_RECORDS_PER_SPLIT);
+ final String clusterId =
DynamicKafkaSourceTestHelper.getKafkaClusterId(0);
+ final KafkaStream stream =
+ new KafkaStream(
+ TOPIC,
+ DynamicKafkaSourceTestHelper.getClusterMetadataMap(0,
TOPIC, topic2));
+
+ final DynamicKafkaSourceEnumState checkpointState =
+ getCheckpointStateWithTopicIntegrityCheck(stream);
+ final KafkaSourceEnumState clusterState =
+ checkpointState.getClusterEnumeratorStates().get(clusterId);
+ final Map<String, String> trackedIds =
clusterState.trackedTopicIdsByName();
+ assertThat(trackedIds).containsOnlyKeys(TOPIC, topic2);
+
+ // a checkpoint taken before the splits of topic2 were handed to a
reader
+ final Set<SplitAndAssignmentStatus> splits =
+ clusterState.splits().stream()
+ .map(
+ splitStatus ->
+
topic2.equals(splitStatus.split().getTopic())
+ ? new SplitAndAssignmentStatus(
+ splitStatus.split(),
+
AssignmentStatus.UNASSIGNED)
+ : splitStatus)
+ .collect(Collectors.toSet());
+ final DynamicKafkaSourceEnumState stateWithUnassignedTopic =
+ withClusterState(
+ checkpointState,
+ clusterId,
+ new KafkaSourceEnumState(
+ splits,
clusterState.initialDiscoveryFinished(), trackedIds));
+
+ try (MockKafkaMetadataService metadataService =
+ new
MockKafkaMetadataService(Collections.singleton(stream));
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ metadataService,
+
DynamicKafkaSourceEnumeratorTest::enableTopicIntegrityCheck,
+ stateWithUnassignedTopic)) {
+ enumerator.start();
+ // the restored sub-enumerator discovers before the first metadata
refresh
+ runAllOneTimeCallables(context);
+
+ assertThat(getTrackedTopicIds(enumerator.snapshotState(-1)))
+ .containsExactlyInAnyOrderEntriesOf(trackedIds);
+ }
+ }
+
+ private static void enableTopicIntegrityCheck(Properties properties) {
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_DISCOVERY_INTERVAL_MS.key(), "1");
+
properties.setProperty(KafkaSourceOptions.TOPIC_INTEGRITY_CHECK_ENABLED.key(),
"true");
+ }
+
+ private DynamicKafkaSourceEnumState
getCheckpointStateWithTopicIntegrityCheck(
+ KafkaStream kafkaStream) throws Throwable {
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStream)),
+
DynamicKafkaSourceEnumeratorTest::enableTopicIntegrityCheck)) {
+ enumerator.start();
+ context.runPeriodicCallable(0);
+ runAllOneTimeCallables(context);
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
0);
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
1);
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
2);
+ return enumerator.snapshotState(-1);
+ }
+ }
+
+ private static DynamicKafkaSourceEnumState withClusterState(
+ DynamicKafkaSourceEnumState state,
+ String clusterId,
+ KafkaSourceEnumState clusterState) {
+ final Map<String, KafkaSourceEnumState> clusterStates =
+ new HashMap<>(state.getClusterEnumeratorStates());
+ clusterStates.put(clusterId, clusterState);
+ return new DynamicKafkaSourceEnumState(state.getKafkaStreams(),
clusterStates);
+ }
+
+ private static Map<String, String>
getTrackedTopicIds(DynamicKafkaSourceEnumState state) {
Review Comment:
If this helper is reused with integrity checking enabled and the default
two-cluster fixture, the same topic name occurs twice and toMap throws
"Duplicate key"?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/StoppableKafkaEnumContextProxy.java:
##########
@@ -276,6 +276,20 @@ protected <T> Callable<T>
wrapCallAsyncCallable(Callable<T> callable) {
};
}
+ /**
+ * The topic integrity check reports a missing or recreated topic as a
{@link
+ * TopicIntegrityException} without the broker exception as cause, so it
has to be matched on
+ * its own to be swallowed for a cluster that is no longer active.
+ */
+ private static Optional<? extends Throwable>
findKafkaRelatedThrowable(Throwable t) {
Review Comment:
Does it make sense to use `ExceptionUtils.findThrowable(Throwable,
Predicate<Throwable>)` ? it can do this, so the helper and the wildcard
`Optional` can go:
```java
Optional<Throwable> throwable =
ExceptionUtils.findThrowable(
e, t -> t instanceof KafkaException || t instanceof
TopicIntegrityException);
```
WDYT?
##########
docs/content/docs/connectors/datastream/dynamic-kafka.md:
##########
@@ -317,6 +317,14 @@ There are configuration options in
DynamicKafkaSourceOptions that can be configu
In addition to this list, see the [regular Kafka connector]({{< ref
"docs/connectors/datastream/kafka" >}}#additional-properties) for
a list of applicable properties.
+### Topic Integrity Check
+Setting the property `scan.topic-integrity-check.enabled` to `true` enables the
Review Comment:
Could we show how to set it, like the KafkaSource page does?
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactory.java:
##########
@@ -161,6 +163,10 @@ public DynamicTableSource createDynamicTableSource(Context
context) {
KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
Long.toString(partitionDiscoveryInterval.toMillis()));
+ properties.setProperty(
Review Comment:
Same
https://github.com/apache/flink-connector-kafka/pull/334/changes#r4427595622
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java:
##########
@@ -222,6 +225,10 @@ public DynamicTableSource createDynamicTableSource(Context
context) {
final String keyPrefix =
tableOptions.getOptional(KEY_FIELDS_PREFIX).orElse(null);
+ properties.setProperty(
Review Comment:
When the top-level option is omitted,
`tableOptions.get(SCAN_TOPIC_INTEGRITY_CHECK_ENABLED)` returns its default
false.
This overwrites an existing 'properties.scan.topic-integrity-check.enabled'
= 'true', and silently disabling checks that were enabled before this change.
Could we apply this override only when the top-level option is explicitly
configured? Explicit false should still take precedence, while an absent option
should preserve the existing setting.
##########
docs/content/docs/connectors/datastream/dynamic-kafka.md:
##########
@@ -317,6 +317,14 @@ There are configuration options in
DynamicKafkaSourceOptions that can be configu
In addition to this list, see the [regular Kafka connector]({{< ref
"docs/connectors/datastream/kafka" >}}#additional-properties) for
a list of applicable properties.
+### Topic Integrity Check
+Setting the property `scan.topic-integrity-check.enabled` to `true` enables the
+[topic integrity check]({{< ref "docs/connectors/datastream/kafka"
>}}#topic-integrity-check) for the topics
+subscribed on every Kafka cluster of the stream: the source fails with a
`TopicIntegrityException` if a
Review Comment:
Do we want an `enableTopicIntegrityCheck()` on `DynamicKafkaSourceBuilder`,
like `KafkaSourceBuilder` has?
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java:
##########
@@ -35,8 +35,10 @@
import org.apache.flink.connector.kafka.source.KafkaSourceOptions;
import org.apache.flink.connector.kafka.source.enumerator.AssignmentStatus;
import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumState;
+import
org.apache.flink.connector.kafka.source.enumerator.SplitAndAssignmentStatus;
Review Comment:
Now DynamicKafkaSourceEnumeratorTest has 3117 lines and exceeds the
Checkstyle limit of 3100.
https://github.com/Efrat19/flink-connector-kafka/actions/runs/40704407963/job/122872517230#step:17:134
Could we move the topic-integrity tests and their helpers into a separate
test class?
--
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]
