mimaison commented on code in PR #11748:
URL: https://github.com/apache/kafka/pull/11748#discussion_r873584393
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java:
##########
@@ -319,27 +319,21 @@ public void testReplication() throws Exception {
waitForCondition(() ->
primaryClient.remoteConsumerOffsets(consumerGroupName, BACKUP_CLUSTER_ALIAS,
Duration.ofMillis(CHECKPOINT_DURATION_MS)).containsKey(new
TopicPartition("backup.test-topic-1", 0)), CHECKPOINT_DURATION_MS, "Offsets not
translated downstream to primary cluster.");
- waitForCondition(() ->
primaryClient.remoteConsumerOffsets(consumerGroupName, BACKUP_CLUSTER_ALIAS,
- Duration.ofMillis(CHECKPOINT_DURATION_MS)).containsKey(new
TopicPartition("test-topic-1", 0)), CHECKPOINT_DURATION_MS, "Offsets not
translated upstream to primary cluster.");
Review Comment:
This was querying checkpoints from the `backup.checkpoints.internal` topic.
This topic is on the primary cluster and contains data emitted by the
backup->primary checkpoint connector.
In the backup->primary mirroring flow, the remote topics (that end up in the
`primary` cluster) are all prefixed with `backup.`. With this change we now
only emit checkpoints for topics present in offset-syncs, so only topics
prefixed with `backup.`.
So the assertion above still works, the checkpoint connector has emitted
checkpoints for `backup.test-topic-1`. But this assertion is not true anymore
for `test-topic-1`. This makes sense because in the primary cluster,
`test-topic-1` is a local topic, so the backup->primary checkpoint connector
has no business committing offsets for that topic.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
