Efrat19 commented on code in PR #267:
URL:
https://github.com/apache/flink-connector-kafka/pull/267#discussion_r3747228103
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaPropertiesUtil.java:
##########
@@ -36,6 +39,41 @@ public static void copyProperties(@Nonnull Properties from,
@Nonnull Properties
}
}
+ /** Resolves an explicit cluster or global reset strategy before the
initializer default. */
+ public static String resolveAutoOffsetResetStrategy(
+ @Nonnull Properties globalProperties,
+ @Nonnull Properties clusterProperties,
+ @Nonnull OffsetsInitializer startingOffsetsInitializer) {
+ String clusterReset =
+
clusterProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (clusterReset != null) {
+ return clusterReset;
+ }
+
+ String globalReset =
globalProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (globalReset != null) {
+ return globalReset;
+ }
+
+ return
startingOffsetsInitializer.getAutoOffsetResetStrategy().name().toLowerCase();
+ }
+
+ /** Returns whether the configured strategy opposes a positional
initializer strategy. */
+ public static boolean hasOpposingOffsetResetStrategies(
+ @Nonnull String configuredResetStrategy,
+ @Nonnull OffsetsInitializer startingOffsetsInitializer) {
+ String initializerResetStrategy =
+ startingOffsetsInitializer.getAutoOffsetResetStrategy().name();
+ return isPositionalResetStrategy(configuredResetStrategy)
+ && isPositionalResetStrategy(initializerResetStrategy)
+ &&
!configuredResetStrategy.equalsIgnoreCase(initializerResetStrategy);
+ }
+
+ private static boolean isPositionalResetStrategy(String resetStrategy) {
+ return "earliest".equalsIgnoreCase(resetStrategy)
+ || "latest".equalsIgnoreCase(resetStrategy);
Review Comment:
`OffsetResetStrategy.LATEST`
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -534,12 +534,27 @@ private KafkaSourceEnumerator
createEnumeratorWithAssignedTopicPartitions(
KafkaPropertiesUtil.copyProperties(properties, consumerProps);
DynamicKafkaSourceOptions.removeRemovedClusterRetentionOption(consumerProps);
KafkaPropertiesUtil.setClientIdPrefix(consumerProps, kafkaClusterId);
+ String effectiveOffsetResetStrategy =
+ KafkaPropertiesUtil.resolveAutoOffsetResetStrategy(
+ properties, fetchedProperties,
effectiveStartingOffsetsInitializer);
+ if (KafkaPropertiesUtil.hasOpposingOffsetResetStrategies(
Review Comment:
I'd move this check to `DynamicKafkaSourceBuilder` with the rest of the
dynamic source config validations, wdyt?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaPropertiesUtil.java:
##########
@@ -36,6 +39,41 @@ public static void copyProperties(@Nonnull Properties from,
@Nonnull Properties
}
}
+ /** Resolves an explicit cluster or global reset strategy before the
initializer default. */
+ public static String resolveAutoOffsetResetStrategy(
+ @Nonnull Properties globalProperties,
+ @Nonnull Properties clusterProperties,
+ @Nonnull OffsetsInitializer startingOffsetsInitializer) {
+ String clusterReset =
+
clusterProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (clusterReset != null) {
+ return clusterReset;
+ }
+
+ String globalReset =
globalProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (globalReset != null) {
+ return globalReset;
+ }
+
+ return
startingOffsetsInitializer.getAutoOffsetResetStrategy().name().toLowerCase();
+ }
+
+ /** Returns whether the configured strategy opposes a positional
initializer strategy. */
+ public static boolean hasOpposingOffsetResetStrategies(
+ @Nonnull String configuredResetStrategy,
+ @Nonnull OffsetsInitializer startingOffsetsInitializer) {
+ String initializerResetStrategy =
+ startingOffsetsInitializer.getAutoOffsetResetStrategy().name();
+ return isPositionalResetStrategy(configuredResetStrategy)
+ && isPositionalResetStrategy(initializerResetStrategy)
+ &&
!configuredResetStrategy.equalsIgnoreCase(initializerResetStrategy);
+ }
+
+ private static boolean isPositionalResetStrategy(String resetStrategy) {
Review Comment:
```suggestion
private static boolean isPositionalResetStrategy(OffsetResetStrategy
resetStrategy) {
```
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaPropertiesUtil.java:
##########
@@ -36,6 +39,41 @@ public static void copyProperties(@Nonnull Properties from,
@Nonnull Properties
}
}
+ /** Resolves an explicit cluster or global reset strategy before the
initializer default. */
+ public static String resolveAutoOffsetResetStrategy(
+ @Nonnull Properties globalProperties,
+ @Nonnull Properties clusterProperties,
+ @Nonnull OffsetsInitializer startingOffsetsInitializer) {
+ String clusterReset =
+
clusterProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (clusterReset != null) {
+ return clusterReset;
+ }
+
+ String globalReset =
globalProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (globalReset != null) {
+ return globalReset;
+ }
+
+ return
startingOffsetsInitializer.getAutoOffsetResetStrategy().name().toLowerCase();
Review Comment:
Nit:
```
return
clusterProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG) ||
globalProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG) ||
startingOffsetsInitializer.getAutoOffsetResetStrategy()
```
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaPropertiesUtilTest.java:
##########
@@ -63,6 +63,31 @@ void
testClusterResetPropertyOverridesGlobalAndInitializerStrategies() {
.isEqualTo("earliest");
}
+ @Test
Review Comment:
I'd switch to `@ParameterizedTest` and cover all 4*4 cases with less code,
wdyt
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -534,12 +534,27 @@ private KafkaSourceEnumerator
createEnumeratorWithAssignedTopicPartitions(
KafkaPropertiesUtil.copyProperties(properties, consumerProps);
DynamicKafkaSourceOptions.removeRemovedClusterRetentionOption(consumerProps);
KafkaPropertiesUtil.setClientIdPrefix(consumerProps, kafkaClusterId);
+ String effectiveOffsetResetStrategy =
Review Comment:
ditto
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java:
##########
@@ -468,10 +468,25 @@ private void parseAndSetRequiredProperties() {
maybeOverride(KafkaSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT.key(), "false",
false);
}
maybeOverride(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false",
false);
+ String configuredOffsetResetStrategy =
Review Comment:
```suggestion
OffsetResetStrategy configuredOffsetResetStrategy =
```
See
[getResetStrategy](https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableSource.java#L513)
(can probably be made a static util)
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaPropertiesUtil.java:
##########
@@ -36,6 +39,41 @@ public static void copyProperties(@Nonnull Properties from,
@Nonnull Properties
}
}
+ /** Resolves an explicit cluster or global reset strategy before the
initializer default. */
+ public static String resolveAutoOffsetResetStrategy(
+ @Nonnull Properties globalProperties,
+ @Nonnull Properties clusterProperties,
+ @Nonnull OffsetsInitializer startingOffsetsInitializer) {
+ String clusterReset =
+
clusterProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (clusterReset != null) {
+ return clusterReset;
+ }
+
+ String globalReset =
globalProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (globalReset != null) {
+ return globalReset;
+ }
+
+ return
startingOffsetsInitializer.getAutoOffsetResetStrategy().name().toLowerCase();
+ }
+
+ /** Returns whether the configured strategy opposes a positional
initializer strategy. */
+ public static boolean hasOpposingOffsetResetStrategies(
+ @Nonnull String configuredResetStrategy,
Review Comment:
```suggestion
@Nonnull OffsetResetStrategy configuredResetStrategy,
```
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaPropertiesUtil.java:
##########
@@ -36,6 +39,41 @@ public static void copyProperties(@Nonnull Properties from,
@Nonnull Properties
}
}
+ /** Resolves an explicit cluster or global reset strategy before the
initializer default. */
+ public static String resolveAutoOffsetResetStrategy(
+ @Nonnull Properties globalProperties,
+ @Nonnull Properties clusterProperties,
+ @Nonnull OffsetsInitializer startingOffsetsInitializer) {
+ String clusterReset =
+
clusterProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (clusterReset != null) {
+ return clusterReset;
+ }
+
+ String globalReset =
globalProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (globalReset != null) {
+ return globalReset;
+ }
+
+ return
startingOffsetsInitializer.getAutoOffsetResetStrategy().name().toLowerCase();
+ }
+
+ /** Returns whether the configured strategy opposes a positional
initializer strategy. */
+ public static boolean hasOpposingOffsetResetStrategies(
+ @Nonnull String configuredResetStrategy,
+ @Nonnull OffsetsInitializer startingOffsetsInitializer) {
+ String initializerResetStrategy =
+ startingOffsetsInitializer.getAutoOffsetResetStrategy().name();
+ return isPositionalResetStrategy(configuredResetStrategy)
+ && isPositionalResetStrategy(initializerResetStrategy)
+ &&
!configuredResetStrategy.equalsIgnoreCase(initializerResetStrategy);
+ }
+
+ private static boolean isPositionalResetStrategy(String resetStrategy) {
+ return "earliest".equalsIgnoreCase(resetStrategy)
Review Comment:
`OffsetResetStrategy.EARLIEST`
--
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]
