seongsu shin created FLINK-41353:
------------------------------------
Summary: [Kafka] KafkaSource rejects
auto.offset.reset=by_duration:<duration> since FLINK-39888
Key: FLINK-41353
URL: https://issues.apache.org/jira/browse/FLINK-41353
Project: Flink
Issue Type: Bug
Components: Connectors / Kafka
Affects Versions: kafka-5.1.0
Reporter: seongsu shin
Kafka 4.0 added a third reset strategy to auto.offset.reset,
by_duration:<ISO-8601 duration> (KIP-1106). The connector already builds
against kafka-clients 4.2.0, which accepts it. Starting with 5.1.0-rc1, though,
KafkaSourceBuilder#build() throws when the property uses this form:
{noformat}
java.lang.IllegalArgumentException: auto.offset.reset can not be set to
by_duration:PT1H. Valid values: [latest,earliest,none]
{noformat}
The check came in with FLINK-39888 (dc707db6).
KafkaSourceBuilder#parseAndSetRequiredProperties now runs the configured value
through KafkaPropertiesUtil#getResetStrategy, which only matches against the
deprecated OffsetResetStrategy enum (EARLIEST, LATEST, NONE). The same helper
is used by DynamicKafkaSourceBuilder and by KafkaDynamicSource /
DynamicKafkaTableSource in group-offsets mode, so the SQL connector should be
affected as well (I only ran the DataStream case).
In 5.0.0 build() does not fail, but only because the configured value was
always overwritten with the initializer's strategy (maybeOverride(..., true)),
so by_duration was never applied there either. Now that 5.1 keeps the user's
value, a value Kafka itself accepts gets rejected, and a job that has this
property set fails at startup after upgrading.
h3. Reproduction
{code:java}
KafkaSource.<String>builder()
.setBootstrapServers("localhost:9092")
.setTopics("t")
.setGroupId("g")
.setStartingOffsets(OffsetsInitializer.committedOffsets())
.setValueOnlyDeserializer(new SimpleStringSchema())
.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"by_duration:PT1H")
.build();
{code}
Creating a plain KafkaConsumer with the same property works. No broker is
needed, it fails inside build().
JDK 21, Flink 2.2.0, kafka-clients 4.2.0:
||connector||new KafkaConsumer(props)||KafkaSource builder build()||
|5.0.0-2.2 (Maven Central)|OK|OK (value is overwritten)|
|5.1.0-2.2 rc1 (staging orgapacheflink-1946)|OK|IllegalArgumentException|
h3. Possible fix
Parse the value with Kafka's own AutoOffsetResetStrategy
(org.apache.kafka.clients.consumer.internals, since 4.0), or pass values the
connector doesn't recognize through and let the consumer's config validation
handle them. The "opposing strategies" warning can just skip by_duration since
it isn't a positional strategy.
I'd be happy to work on a fix if this sounds reasonable.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
