mimaison commented on code in PR #11748:
URL: https://github.com/apache/kafka/pull/11748#discussion_r873670168
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java:
##########
@@ -526,6 +520,65 @@ public void testOffsetSyncsTopicsOnTarget() throws
Exception {
assertFalse(primaryTopics.contains("mm2-offset-syncs." +
BACKUP_CLUSTER_ALIAS + ".internal"));
}
+ @Test
+ public void testNoCheckpointsIfNoRecordsAreMirrored() throws
InterruptedException {
+ String consumerGroupName = "consumer-group-no-checkpoints";
+ Map<String, Object> consumerProps =
Collections.singletonMap("group.id", consumerGroupName);
+
+ // ensure there are some records in the topic on the source cluster
+ produceMessages(primary, "test-topic-1");
+
+ // warm up consumers before starting the connectors, so we don't need
to wait for discovery
+ warmUpConsumer(consumerProps);
+
+ // one way replication from primary to backup
+ mm2Props.put(BACKUP_CLUSTER_ALIAS + "->" + PRIMARY_CLUSTER_ALIAS +
".enabled", "false");
+ mm2Config = new MirrorMakerConfig(mm2Props);
+ waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config,
PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);
+
+ // make sure the topics are created in the backup cluster
+ waitForTopicCreated(backup, remoteTopicName("test-topic-1",
PRIMARY_CLUSTER_ALIAS));
+ waitForTopicCreated(backup, remoteTopicName("test-topic-no-records",
PRIMARY_CLUSTER_ALIAS));
+
+ // commit some offsets for both topics in the source cluster
+ TopicPartition tp1 = new TopicPartition("test-topic-1", 0);
+ TopicPartition tp2 = new TopicPartition("test-topic-no-records", 0);
+ //Map<String, Object> consumerProps =
Collections.singletonMap("group.id", consumerGroupName);
+ try (Consumer<byte[], byte[]> consumer =
primary.kafka().createConsumer(consumerProps)) {
+ Collection<TopicPartition> tps = Arrays.asList(tp1, tp2);
+ Map<TopicPartition, Long> endOffsets = consumer.endOffsets(tps);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
endOffsets.entrySet().stream()
+ .collect(Collectors.toMap(
+ Map.Entry::getKey,
+ e -> new OffsetAndMetadata(e.getValue())
+ ));
+ consumer.commitSync(offsetsToCommit);
+ }
+
+ // Only test-topic-1 should have translated offsets because we've not
yet mirrored any records for topic-no-records
+ MirrorClient backupClient = new
MirrorClient(mm2Config.clientConfig(BACKUP_CLUSTER_ALIAS));
+ waitForCondition(() -> {
+ Map<TopicPartition, OffsetAndMetadata> translatedOffsets =
backupClient.remoteConsumerOffsets(
+ consumerGroupName, PRIMARY_CLUSTER_ALIAS,
Duration.ofSeconds(30L));
+ return translatedOffsets.containsKey(remoteTopicPartition(tp1,
PRIMARY_CLUSTER_ALIAS)) &&
+ !translatedOffsets.containsKey(remoteTopicPartition(tp2,
PRIMARY_CLUSTER_ALIAS));
+ }, OFFSET_SYNC_DURATION_MS, "Checkpoints were not emitted correctly to
backup cluster");
+
+ // Send some records to topic-no-records in the source cluster
Review Comment:
I've renamed to `test-topic-no-checkpoints` so it match the test/group names
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
