You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2026/06/01 19:23:43 UTC
(flink-connector-kafka) branch main updated: [FLINK-39723][tests] Wait for partition assignment in KafkaTableTestBase.createTestTopic
This is an automated email from the ASF dual-hosted git repository.
MartijnVisser 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 84d2ee7a [FLINK-39723][tests] Wait for partition assignment in KafkaTableTestBase.createTestTopic
84d2ee7a is described below
commit 84d2ee7a53c892ed21d4d6ddadcd49322571b957
Author: Aleksandr Savonin <sa...@gmail.com>
AuthorDate: Mon Jun 1 20:20:15 2026 +0200
[FLINK-39723][tests] Wait for partition assignment in KafkaTableTestBase.createTestTopic
---
.../connectors/kafka/table/KafkaTableTestBase.java | 18 +++---------------
1 file changed, 3 insertions(+), 15 deletions(-)
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
index 5db67f6a..32e76f68 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
@@ -29,7 +29,6 @@ import org.apache.flink.test.util.AbstractTestBase;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
-import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.RecordsToDelete;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.TopicListing;
@@ -122,21 +121,10 @@ abstract class KafkaTableTestBase extends AbstractTestBase {
}
public void createTestTopic(String topic, int numPartitions, int replicationFactor) {
- Map<String, Object> properties = new HashMap<>();
+ Properties properties = new Properties();
properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
- try (AdminClient admin = AdminClient.create(properties)) {
- admin.createTopics(
- Collections.singletonList(
- new NewTopic(topic, numPartitions, (short) replicationFactor)))
- .all()
- .get();
- } catch (Exception e) {
- throw new IllegalStateException(
- String.format(
- "Fail to create topic [%s partitions: %d replication factor: %d].",
- topic, numPartitions, replicationFactor),
- e);
- }
+ KafkaUtil.createNewTopicAndWaitForPartitionAssignment(
+ topic, numPartitions, replicationFactor, properties);
}
public Map<TopicPartition, OffsetAndMetadata> getConsumerOffset(String groupId) {