(paimon) branch master updated: [e2e] Fix kafka consumer concurrent exception (#4261)

[lzljs3620320]

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

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 722994eba [e2e] Fix kafka consumer concurrent exception (#4261)
722994eba is described below

commit 722994eba11e086f44c0a836afba857bd9e11cdc
Author: Jie Feng <shadow...@126.com>
AuthorDate: Thu Sep 26 15:57:49 2024 +0800

    [e2e] Fix kafka consumer concurrent exception (#4261)
---
 .../src/test/java/org/apache/paimon/tests/E2eTestBase.java             | 2 +-
 .../apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java    | 3 ++-
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git 
a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java 
b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java
index 738330aee..55f9e984e 100644
--- a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java
+++ b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java
@@ -190,7 +190,7 @@ public abstract class E2eTestBase {
                 "cat >" + TEST_DATA_DIR + "/" + filename + " <<EOF\n" + 
content + "EOF\n");
     }
 
-    protected void createKafkaTopic(String topicName, int partitionNum)
+    protected synchronized void createKafkaTopic(String topicName, int 
partitionNum)
             throws IOException, InterruptedException {
         assert withKafka;
         ContainerState kafka = 
environment.getContainerByServiceName("kafka-1").get();
diff --git 
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
 
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
index 8f41a370e..8db2e7ffe 100644
--- 
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
+++ 
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
@@ -227,7 +227,8 @@ public abstract class KafkaActionITCaseBase extends 
CdcActionITCaseBase {
         }
     }
 
-    private void logTopicPartitionStatus(Map<String, TopicDescription> 
topicDescriptions) {
+    private synchronized void logTopicPartitionStatus(
+            Map<String, TopicDescription> topicDescriptions) {
         List<TopicPartition> partitions = new ArrayList<>();
         topicDescriptions.forEach(
                 (topic, description) ->