jigar-bhati commented on code in PR #267:
URL:
https://github.com/apache/flink-connector-kafka/pull/267#discussion_r3626596472
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java:
##########
@@ -283,7 +272,10 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
clusterMetadataMapEntry.getValue().getProperties(),
clusterProperties);
OffsetsInitializer startingOffsetsInitializer =
clusterMetadataMapEntry.getValue().getStartingOffsetsInitializer();
- if (startingOffsetsInitializer != null) {
+ if (startingOffsetsInitializer != null
+ &&
!properties.containsKey(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)
Review Comment:
Good catch. The dynamic builder no longer materializes an
initializer-derived when it is absent. The reader now receives the global
initializer and resolves explicit global > explicit cluster > effective
initializer, so a cluster-specific initializer is still honored when no reset
property is configured.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -601,12 +534,13 @@ private KafkaSourceEnumerator
createEnumeratorWithAssignedTopicPartitions(
KafkaPropertiesUtil.copyProperties(properties, consumerProps);
DynamicKafkaSourceOptions.removeRemovedClusterRetentionOption(consumerProps);
KafkaPropertiesUtil.setClientIdPrefix(consumerProps, kafkaClusterId);
- consumerProps.setProperty(
+ consumerProps.computeIfAbsent(
Review Comment:
Yes, this was the same issue. The enumerator now uses the shared
reset-strategy resolver with the raw global properties, fetched cluster
properties, and effective initializer, so a generated default can no longer
mask the cluster initializer and reader/enumerator precedence stays consistent.
--
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]
