protected static Map<TopicPartition, OffsetAndMetadata> waitForNewCheckpointOnAllPartitions(
                MirrorClient client, String consumerGroupName, String remoteClusterAlias, String topicName,
                Map<TopicPartition, OffsetAndMetadata> lastCheckpoint
    ) throws InterruptedException {
        AtomicReference<Map<TopicPartition, OffsetAndMetadata>> ret = new AtomicReference<>();
        waitForCondition(
                () -> {
                    Map<TopicPartition, OffsetAndMetadata> offsets = client.remoteConsumerOffsets(
                            consumerGroupName, remoteClusterAlias, Duration.ofMillis(3000));
                    for (int i = 0; i < NUM_PARTITIONS; i++) {
                        TopicPartition tp = new TopicPartition(topicName, i);
                        if (!offsets.containsKey(tp)) {
                            log.info("Checkpoint is missing for {}: {}-{}", consumerGroupName, topicName, i);
                            return false;
                        } else if (lastCheckpoint.containsKey(tp) && lastCheckpoint.get(tp).equals(offsets.get(tp))) {
                            log.info("Checkpoint is the same as previous checkpoint");
                            return false;
                        }
                    }
                    ret.set(offsets);
                    return true;
                },
                CHECKPOINT_DURATION_MS,
                String.format(
                        "Offsets for consumer group %s not translated from %s for topic %s",
                        consumerGroupName,
                        remoteClusterAlias,
                        topicName
                )
        );
        return ret.get();
    }
Produce a batch of records with the specified producer
@param producer Producer to use to send records
@param records Records to send in one parallel batch
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java