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 f2cc850a4f07d..7aae7d51eb25b 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 @@ -47,6 +47,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Properties; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -82,6 +83,7 @@ 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 NUM_WORKERS = 3; private static final Duration CONSUMER_POLL_TIMEOUT_MS = Duration.ofMillis(500); protected static final String PRIMARY_CLUSTER_ALIAS = "primary"; @@ -227,7 +229,10 @@ public void testReplication() throws Exception { MirrorClient primaryClient = new MirrorClient(mm2Config.clientConfig(PRIMARY_CLUSTER_ALIAS)); MirrorClient backupClient = new MirrorClient(mm2Config.clientConfig(BACKUP_CLUSTER_ALIAS)); - + + // make sure the topic is auto-created in the other cluster + waitForTopicCreated(primary.kafka(), "backup.test-topic-1"); + waitForTopicCreated(backup.kafka(), "primary.test-topic-1"); assertEquals(TopicConfig.CLEANUP_POLICY_COMPACT, getTopicConfig(backup.kafka(), "primary.test-topic-1", TopicConfig.CLEANUP_POLICY_CONFIG), "topic config was not synced"); @@ -312,6 +317,10 @@ public void testReplication() throws Exception { primary.kafka().createTopic("test-topic-2", NUM_PARTITIONS); backup.kafka().createTopic("test-topic-3", NUM_PARTITIONS); + // make sure the topic is auto-created in the other cluster + waitForTopicCreated(backup.kafka(), "primary.test-topic-2"); + waitForTopicCreated(primary.kafka(), "backup.test-topic-3"); + // only produce messages to the first partition produceMessages(primary, "test-topic-2", 1); produceMessages(backup, "test-topic-3", 1); @@ -360,15 +369,17 @@ public void testReplicationWithEmptyPartition() throws Exception { waitForConsumingAllRecords(backupConsumer, expectedRecords); } - Admin backupClient = backup.kafka().createAdminClient(); - // retrieve the consumer group offset from backup cluster - Map remoteOffsets = + try (Admin backupClient = backup.kafka().createAdminClient()) { + // retrieve the consumer group offset from backup cluster + Map remoteOffsets = backupClient.listConsumerGroupOffsets(consumerGroupName).partitionsToOffsetAndMetadata().get(); - // pinpoint the offset of the last partition which does not receive records - OffsetAndMetadata offset = remoteOffsets.get(new TopicPartition(PRIMARY_CLUSTER_ALIAS + "." + topic, NUM_PARTITIONS - 1)); - // offset of the last partition should exist, but its value should be 0 - assertNotNull(offset, "Offset of last partition was not replicated"); - assertEquals(0, offset.offset(), "Offset of last partition is not zero"); + + // pinpoint the offset of the last partition which does not receive records + OffsetAndMetadata offset = remoteOffsets.get(new TopicPartition(PRIMARY_CLUSTER_ALIAS + "." + topic, NUM_PARTITIONS - 1)); + // offset of the last partition should exist, but its value should be 0 + assertNotNull(offset, "Offset of last partition was not replicated"); + assertEquals(0, offset.offset(), "Offset of last partition is not zero"); + } } @Test @@ -396,6 +407,9 @@ public void testOneWayReplicationWithAutoOffsetSync() throws InterruptedExceptio waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS); + // make sure the topic is created in the other cluster + waitForTopicCreated(primary.kafka(), "backup.test-topic-1"); + waitForTopicCreated(backup.kafka(), "primary.test-topic-1"); // create a consumer at backup cluster with same consumer group Id to consume 1 topic Consumer backupConsumer = backup.kafka().createConsumerAndSubscribeTo( consumerProps, "primary.test-topic-1"); @@ -412,7 +426,9 @@ public void testOneWayReplicationWithAutoOffsetSync() throws InterruptedExceptio // now create a new topic in primary cluster primary.kafka().createTopic("test-topic-2", NUM_PARTITIONS); - backup.kafka().createTopic("primary.test-topic-2", 1); + // make sure the topic is created in backup cluster + waitForTopicCreated(backup.kafka(), "primary.test-topic-2"); + // produce some records to the new topic in primary cluster produceMessages(primary, "test-topic-2"); @@ -455,27 +471,41 @@ private static void waitUntilMirrorMakerIsRunning(EmbeddedConnectCluster connect "Connector " + connector.getSimpleName() + " tasks did not start in time on cluster: " + connectCluster); } } - + + /* + * wait for the topic created on the cluster + */ + private static void waitForTopicCreated(EmbeddedKafkaCluster cluster, String topicName) throws InterruptedException { + try (final Admin adminClient = cluster.createAdminClient()) { + waitForCondition(() -> adminClient.listTopics().names().get().contains(topicName), TOPIC_SYNC_DURATION_MS, + "Topic: " + topicName + " didn't get created in the cluster" + ); + } + } + /* * delete all topics of the input kafka cluster */ private static void deleteAllTopics(EmbeddedKafkaCluster cluster) throws Exception { - Admin client = cluster.createAdminClient(); - client.deleteTopics(client.listTopics().names().get()); + try (final Admin adminClient = cluster.createAdminClient()) { + Set topicsToBeDeleted = adminClient.listTopics().names().get(); + log.debug("Deleting topics: {} ", topicsToBeDeleted); + adminClient.deleteTopics(topicsToBeDeleted).all().get(); + } } /* * retrieve the config value based on the input cluster, topic and config name */ - private static String getTopicConfig(EmbeddedKafkaCluster cluster, String topic, String configName) - throws Exception { - Admin client = cluster.createAdminClient(); - Collection cr = Collections.singleton( - new ConfigResource(ConfigResource.Type.TOPIC, topic)); - - DescribeConfigsResult configsResult = client.describeConfigs(cr); - Config allConfigs = (Config) configsResult.all().get().values().toArray()[0]; - return allConfigs.get(configName).value(); + private static String getTopicConfig(EmbeddedKafkaCluster cluster, String topic, String configName) throws Exception { + try (Admin client = cluster.createAdminClient()) { + Collection cr = Collections.singleton( + new ConfigResource(ConfigResource.Type.TOPIC, topic)); + + DescribeConfigsResult configsResult = client.describeConfigs(cr); + Config allConfigs = (Config) configsResult.all().get().values().toArray()[0]; + return allConfigs.get(configName).value(); + } } /* @@ -505,27 +535,28 @@ protected void produceMessages(EmbeddedConnectCluster cluster, String topicName, private static void waitForConsumerGroupOffsetSync(EmbeddedConnectCluster connect, Consumer consumer, List topics, String consumerGroupId, int numRecords) throws InterruptedException { - Admin adminClient = connect.kafka().createAdminClient(); - List tps = new ArrayList<>(NUM_PARTITIONS * topics.size()); - for (int partitionIndex = 0; partitionIndex < NUM_PARTITIONS; partitionIndex++) { - for (String topic : topics) { - tps.add(new TopicPartition(topic, partitionIndex)); + try (Admin adminClient = connect.kafka().createAdminClient()) { + List tps = new ArrayList<>(NUM_PARTITIONS * topics.size()); + for (int partitionIndex = 0; partitionIndex < NUM_PARTITIONS; partitionIndex++) { + for (String topic : topics) { + tps.add(new TopicPartition(topic, partitionIndex)); + } } - } - long expectedTotalOffsets = numRecords * topics.size(); + long expectedTotalOffsets = numRecords * topics.size(); - waitForCondition(() -> { - Map consumerGroupOffsets = + waitForCondition(() -> { + Map consumerGroupOffsets = adminClient.listConsumerGroupOffsets(consumerGroupId).partitionsToOffsetAndMetadata().get(); - long consumerGroupOffsetTotal = consumerGroupOffsets.values().stream() + long consumerGroupOffsetTotal = consumerGroupOffsets.values().stream() .mapToLong(OffsetAndMetadata::offset).sum(); - Map offsets = consumer.endOffsets(tps, CONSUMER_POLL_TIMEOUT_MS); - long totalOffsets = offsets.values().stream().mapToLong(l -> l).sum(); + Map offsets = consumer.endOffsets(tps, CONSUMER_POLL_TIMEOUT_MS); + long totalOffsets = offsets.values().stream().mapToLong(l -> l).sum(); - // make sure the consumer group offsets are synced to expected number - return totalOffsets == expectedTotalOffsets && consumerGroupOffsetTotal > 0; - }, OFFSET_SYNC_DURATION_MS, "Consumer group offset sync is not complete in time"); + // make sure the consumer group offsets are synced to expected number + return totalOffsets == expectedTotalOffsets && consumerGroupOffsetTotal > 0; + }, OFFSET_SYNC_DURATION_MS, "Consumer group offset sync is not complete in time"); + } } /*