bowenli86 commented on code in PR #267:
URL:
https://github.com/apache/flink-connector-kafka/pull/267#discussion_r3799782664
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java:
##########
@@ -468,10 +469,30 @@ private void parseAndSetRequiredProperties() {
maybeOverride(KafkaSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT.key(), "false",
false);
}
maybeOverride(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false",
false);
+ String configuredOffsetReset =
props.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+ if (configuredOffsetReset != null) {
+ OffsetResetStrategy configuredOffsetResetStrategy =
+
KafkaPropertiesUtil.getResetStrategy(configuredOffsetReset);
Review Comment:
This accepts values case-insensitively via `getResetStrategy(...)`, but
leaves the original string in `props` because the later `maybeOverride(...,
false)` preserves explicit values. For example, `auto.offset.reset=EARLIEST`
builds successfully and then reaches Kafka unchanged, where the consumer only
accepts lowercase `earliest`, `latest`, or `none`. Could we normalize the
validated value here (or reject non-lowercase values early) and add a
static-builder regression test, matching the Table path?
--
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]
