Aleksandr Savonin created FLINK-40847:
-----------------------------------------
Summary: KafkaDynamicSink#equals/hashCode ignore
transactionNamingStrategy
Key: FLINK-40847
URL: https://issues.apache.org/jira/browse/FLINK-40847
Project: Flink
Issue Type: Bug
Components: Connectors / Kafka
Reporter: Aleksandr Savonin
KafkaDynamicSink.equals() and .hashCode() do not include the field
transactionNamingStrategy.
The field, together with the table option sink.transaction-naming-strategy, was
introduced before. That change added the field to the constructor, to copy()
and to the KafkaSinkBuilder call, but not to equals/hashCode, which still
compare every other configuration field of the sink (delivery guarantee,
transactional id prefix, topics, partitioner, flush mode, parallelism, ...).
KafkaDynamicSource compares all of its configuration fields.
Impact:
* Two KafkaDynamicSink instances that differ only in the naming strategy
(INCREMENTING vs POOLING) compare equal.
* KafkaDynamicTableFactoryTest#testTableSinkStrategyTranslation and
UpsertKafkaDynamicTableFactoryTest#testTableSinkStrategyTranslation assert
actualSink.isEqualTo(expectedSink). Because equals ignores the strategy, these
tests pass for any value, so a regression in the option translation (e.g. the
factory passing TransactionNamingStrategy.DEFAULT instead of the configured
value) would not be detected.
* No runtime impact: the planner does not compare DynamicTableSink instances,
and the runtime sink is built from the field directly in
getSinkRuntimeProvider. The sink is @Internal.
Fix: add transactionNamingStrategy to equals and hashCode. Optionally
strengthen the two translation tests by asserting that the actual sink is not
equal to an expected sink built with a different strategy, which fails without
the fix and passes with it.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
