You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2026/10/01 20:48:05 UTC
(flink-connector-kafka) branch main updated: [FLINK-40847][Connectors/Kafka] Include transaction naming strategy in sink equality (#330)
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 <sa...@gmail.com>
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);
+ }
+ }
}
}