This is an automated email from the ASF dual-hosted git repository.
lihaosky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new c996f94b [FLINK-40847][Connectors/Kafka] Include transaction naming
strategy in sink equality (#330)
c996f94b is described below
commit c996f94b8c03e5d25ed981660e2d0c18f74e1739
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Thu Oct 1 22:47:59 2026 +0200
[FLINK-40847][Connectors/Kafka] Include transaction naming strategy in sink
equality (#330)
* [FLINK-40847][Connectors/Kafka] Include transaction naming strategy in
sink equality
Include transactionNamingStrategy in KafkaDynamicSink.equals and hashCode.
Extend both factory tests to distinguish strategies using independent
option maps and verify equality and hash-code consistency for copied sinks.
Generated-by: Codex (GPT-6 Astra)
* [FLINK-40847][Connectors/Kafka] Address sink equality review feedback
Compare transaction naming strategies before more expensive fields and
use AssertJ's hasSameHashCodeAs in both factory tests.
Generated-by: Codex (GPT-6 Astra)
---
.../streaming/connectors/kafka/table/KafkaDynamicSink.java | 4 +++-
.../kafka/table/KafkaDynamicTableFactoryTest.java | 13 +++++++++++++
.../kafka/table/UpsertKafkaDynamicTableFactoryTest.java | 13 +++++++++++++
3 files changed, 29 insertions(+), 1 deletion(-)
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
index 9c6c0fb2..4e04329e 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
@@ -325,7 +325,8 @@ public class KafkaDynamicSink implements DynamicTableSink,
SupportsWritingMetada
return false;
}
final KafkaDynamicSink that = (KafkaDynamicSink) o;
- return Objects.equals(metadataKeys, that.metadataKeys)
+ return Objects.equals(transactionNamingStrategy,
that.transactionNamingStrategy)
+ && Objects.equals(metadataKeys, that.metadataKeys)
&& Objects.equals(consumedDataType, that.consumedDataType)
&& Objects.equals(physicalDataType, that.physicalDataType)
&& Objects.equals(keyEncodingFormat, that.keyEncodingFormat)
@@ -363,6 +364,7 @@ public class KafkaDynamicSink implements DynamicTableSink,
SupportsWritingMetada
upsertMode,
flushMode,
transactionalIdPrefix,
+ transactionNamingStrategy,
parallelism);
}
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java
index 63cc1b24..043de12a 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java
@@ -832,6 +832,19 @@ class KafkaDynamicTableFactoryTest {
"kafka-sink",
namingStrategy);
assertThat(actualSink).isEqualTo(expectedSink);
+ final DynamicTableSink copiedSink = actualSink.copy();
+ assertThat(copiedSink).isEqualTo(actualSink);
+ assertThat(copiedSink).hasSameHashCodeAs(actualSink);
+
+ for (TransactionNamingStrategy otherStrategy :
TransactionNamingStrategy.values()) {
+ if (otherStrategy != namingStrategy) {
+ final Map<String, String> otherOptions = new
HashMap<>(modifiedOptions);
+ otherOptions.put("sink.transaction-naming-strategy",
otherStrategy.name());
+ assertThat(createTableSink(SCHEMA, otherOptions))
+ .as("Sinks using %s and %s", otherStrategy,
namingStrategy)
+ .isNotEqualTo(actualSink);
+ }
+ }
}
}
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java
index 8832e428..3e4f9d8c 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaDynamicTableFactoryTest.java
@@ -459,6 +459,19 @@ class UpsertKafkaDynamicTableFactoryTest {
"kafka-sink",
namingStrategy);
assertThat(actualSink).isEqualTo(expectedSink);
+ final DynamicTableSink copiedSink = actualSink.copy();
+ assertThat(copiedSink).isEqualTo(actualSink);
+ assertThat(copiedSink).hasSameHashCodeAs(actualSink);
+
+ for (TransactionNamingStrategy otherStrategy :
TransactionNamingStrategy.values()) {
+ if (otherStrategy != namingStrategy) {
+ final Map<String, String> otherOptions = new
HashMap<>(modifiedOptions);
+ otherOptions.put("sink.transaction-naming-strategy",
otherStrategy.name());
+ assertThat(createTableSink(SINK_SCHEMA, otherOptions))
+ .as("Sinks using %s and %s", otherStrategy,
namingStrategy)
+ .isNotEqualTo(actualSink);
+ }
+ }
}
}
