From 5c5b584c1a737584cbb7fc8887308f5915b16e5f Mon Sep 17 00:00:00 2001 From: Luke Chen Date: Thu, 15 Apr 2021 14:58:43 +0800 Subject: [PATCH 1/3] KAFKA-12284: increase request timeout to make tests reliable --- .../MirrorConnectorsIntegrationBaseTest.java | 30 +++++++++++++----- .../util/clusters/EmbeddedKafkaCluster.java | 31 +++++++++++++------ 2 files changed, 45 insertions(+), 16 deletions(-) diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java index e6bdb9679d64d..841eedc33c6e6 100644 --- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java @@ -17,6 +17,7 @@ package org.apache.kafka.connect.mirror.integration; import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.AdminClientConfig; import org.apache.kafka.clients.admin.Config; import org.apache.kafka.clients.admin.DescribeConfigsResult; import org.apache.kafka.clients.consumer.Consumer; @@ -83,7 +84,8 @@ public abstract class MirrorConnectorsIntegrationBaseTest { private static final int CHECKPOINT_DURATION_MS = 20_000; private static final int RECORD_CONSUME_DURATION_MS = 20_000; private static final int OFFSET_SYNC_DURATION_MS = 30_000; - private static final int TOPIC_SYNC_DURATION_MS = 30_000; + private static final int TOPIC_SYNC_DURATION_MS = 60_000; + private static final int REQUEST_TIMEOUT_DURATION_MS = 60_000; private static final int NUM_WORKERS = 3; private static final Duration CONSUMER_POLL_TIMEOUT_MS = Duration.ofMillis(500); protected static final String PRIMARY_CLUSTER_ALIAS = "primary"; @@ -165,11 +167,19 @@ public void startClusters() throws Exception { primary.start(); primary.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS, "Workers of " + PRIMARY_CLUSTER_ALIAS + "-connect-cluster did not start in time."); + + waitForTopicCreated(primary, "mm2-status.backup.internal"); + waitForTopicCreated(primary, "mm2-offsets.backup.internal"); + waitForTopicCreated(primary, "mm2-configs.backup.internal"); backup.start(); backup.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS, "Workers of " + BACKUP_CLUSTER_ALIAS + "-connect-cluster did not start in time."); + waitForTopicCreated(backup, "mm2-status.primary.internal"); + waitForTopicCreated(backup, "mm2-offsets.primary.internal"); + waitForTopicCreated(backup, "mm2-configs.primary.internal"); + createTopics(); warmUpConsumer(Collections.singletonMap("group.id", "consumer-group-dummy")); @@ -602,13 +612,19 @@ private static Map basicMM2Config() { private void createTopics() { // to verify topic config will be sync-ed across clusters Map topicConfig = Collections.singletonMap(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT); + Map emptyMap = Collections.emptyMap(); + + // increase admin client request timeout value to make the tests reliable. + Properties adminClientConfig = new Properties(); + adminClientConfig.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, REQUEST_TIMEOUT_DURATION_MS); + // create these topics before starting the connectors so we don't need to wait for discovery - primary.kafka().createTopic("test-topic-1", NUM_PARTITIONS, 1, topicConfig); - primary.kafka().createTopic("backup.test-topic-1", 1); - primary.kafka().createTopic("heartbeats", 1); - backup.kafka().createTopic("test-topic-1", NUM_PARTITIONS); - backup.kafka().createTopic("primary.test-topic-1", 1); - backup.kafka().createTopic("heartbeats", 1); + primary.kafka().createTopic("test-topic-1", NUM_PARTITIONS, 1, topicConfig, adminClientConfig); + primary.kafka().createTopic("backup.test-topic-1", 1, 1, emptyMap, adminClientConfig); + primary.kafka().createTopic("heartbeats", 1, 1, emptyMap, adminClientConfig); + backup.kafka().createTopic("test-topic-1", NUM_PARTITIONS, 1, emptyMap, adminClientConfig); + backup.kafka().createTopic("primary.test-topic-1", 1, 1, emptyMap, adminClientConfig); + backup.kafka().createTopic("heartbeats", 1, 1, emptyMap, adminClientConfig); } /* diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java index a8328c5069460..eb608e653fd58 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java @@ -337,18 +337,28 @@ public void createTopic(String topic) { * @param topic The name of the topic. */ public void createTopic(String topic, int partitions) { - createTopic(topic, partitions, 1, new HashMap<>()); + createTopic(topic, partitions, 1, new HashMap<>(), new Properties()); } /** - * Create a Kafka topic with the given parameters. + * Create a Kafka topic with given partition, replication factor, and topic config. * - * @param topic The name of the topic. - * @param partitions The number of partitions for this topic. - * @param replication The replication factor for (partitions of) this topic. - * @param topicConfig Additional topic-level configuration settings. + * @param topic The name of the topic. */ public void createTopic(String topic, int partitions, int replication, Map topicConfig) { + createTopic(topic, partitions, replication, topicConfig, new Properties()); + } + + /** + * Create a Kafka topic with the given parameters. + * + * @param topic The name of the topic. + * @param partitions The number of partitions for this topic. + * @param replication The replication factor for (partitions of) this topic. + * @param topicConfig Additional topic-level configuration settings. + * @param adminClientConfig Additional admin client configuration settings. + */ + public void createTopic(String topic, int partitions, int replication, Map topicConfig, Properties adminClientConfig) { if (replication > brokers.length) { throw new InvalidReplicationFactorException("Insufficient brokers (" + brokers.length + ") for desired replication (" + replication + ")"); @@ -359,7 +369,7 @@ public void createTopic(String topic, int partitions, int replication, Map Date: Wed, 28 Apr 2021 10:10:23 +0800 Subject: [PATCH 2/3] KAFKA-12284: refactor code --- .../kafka/connect/util/clusters/EmbeddedKafkaCluster.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java index eb608e653fd58..49b8a67cb68f3 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java @@ -337,7 +337,7 @@ public void createTopic(String topic) { * @param topic The name of the topic. */ public void createTopic(String topic, int partitions) { - createTopic(topic, partitions, 1, new HashMap<>(), new Properties()); + createTopic(topic, partitions, 1, Collections.emptyMap(), new Properties()); } /** From 02b401a7b7b2641da8b617a99e3edee43df93aa1 Mon Sep 17 00:00:00 2001 From: Luke Chen Date: Wed, 28 Apr 2021 11:41:14 +0800 Subject: [PATCH 3/3] KAFKA-12284: refactor --- .../kafka/connect/util/clusters/EmbeddedKafkaCluster.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java index 49b8a67cb68f3..30d41d26a56e2 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java @@ -337,7 +337,7 @@ public void createTopic(String topic) { * @param topic The name of the topic. */ public void createTopic(String topic, int partitions) { - createTopic(topic, partitions, 1, Collections.emptyMap(), new Properties()); + createTopic(topic, partitions, 1, Collections.emptyMap()); } /**