(flink-connector-kafka) branch main updated: [FLINK-39338] Fix topic name collisions in KafkaWriterTestBase for parameterized tests

[fanrui]

This is an automated email from the ASF dual-hosted git repository.

fanrui 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 bd2fd2b2 [FLINK-39338] Fix topic name collisions in 
KafkaWriterTestBase for parameterized tests
bd2fd2b2 is described below

commit bd2fd2b2c8a48b7d2a05387fc4fda61e878c902f
Author: Aleksandr Savonin <savoni...@gmail.com>
AuthorDate: Thu Mar 26 19:12:42 2026 +0100

    [FLINK-39338] Fix topic name collisions in KafkaWriterTestBase for 
parameterized tests
    
    Fix topic name collision in KafkaWriterTestBase for parameterized tests
---
 .../org/apache/flink/connector/kafka/sink/KafkaWriterTestBase.java | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterTestBase.java
 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterTestBase.java
index 16ae534d..a29e45d6 100644
--- 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterTestBase.java
+++ 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterTestBase.java
@@ -54,6 +54,7 @@ import org.testcontainers.junit.jupiter.Testcontainers;
 import javax.annotation.Nullable;
 
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -98,7 +99,11 @@ public abstract class KafkaWriterTestBase {
     public void setUp(TestInfo testInfo) {
         metricListener = new MetricListener();
         timeService = new TriggerTimeService();
-        topic = testInfo.getDisplayName().replaceAll("\\W", "");
+        String methodName = 
testInfo.getTestMethod().map(Method::getName).orElse("unknown");
+        String displayName = testInfo.getDisplayName().replaceAll("\\W", "");
+        // Use the display name if it already contains the method name,
+        // otherwise combine them to ensure uniqueness for parameterized tests.
+        topic = displayName.startsWith(methodName) ? displayName : methodName 
+ "_" + displayName;
         Map<String, Object> properties = new java.util.HashMap<>();
         properties.put(BOOTSTRAP_SERVERS_CONFIG, 
KAFKA_CONTAINER.getBootstrapServers());
         try (Admin admin = AdminClient.create(properties)) {