private static void assertMonotonicCheckpoints(EmbeddedConnectCluster cluster, String checkpointTopic) {
        TopicPartition checkpointTopicPartition = new TopicPartition(checkpointTopic, 0);
        try (Consumer<byte[], byte[]> backupConsumer = cluster.kafka().createConsumerAndSubscribeTo(Collections.singletonMap(
                "auto.offset.reset", "earliest"), checkpointTopic)) {
            Map<String, Map<TopicPartition, Checkpoint>> checkpointsByGroup = new HashMap<>();
            long deadline = System.currentTimeMillis() + CHECKPOINT_DURATION_MS;
            do {
                ConsumerRecords<byte[], byte[]> records = backupConsumer.poll(Duration.ofSeconds(1L));
                for (ConsumerRecord<byte[], byte[]> record : records) {
                    Checkpoint checkpoint = Checkpoint.deserializeRecord(record);
                    Map<TopicPartition, Checkpoint> lastCheckpoints = checkpointsByGroup.computeIfAbsent(
                            checkpoint.consumerGroupId(),
                            ignored -> new HashMap<>());
                    Checkpoint lastCheckpoint = lastCheckpoints.getOrDefault(checkpoint.topicPartition(), checkpoint);
                    assertTrue(checkpoint.downstreamOffset() >= lastCheckpoint.downstreamOffset(),
                            "Checkpoint was non-monotonic for "
                                    + checkpoint.consumerGroupId()
                                    + ": "
                                    + checkpoint.topicPartition());
                    lastCheckpoints.put(checkpoint.topicPartition(), checkpoint);
                }
            } while (backupConsumer.currentLag(checkpointTopicPartition).orElse(1) > 0 && System.currentTimeMillis() < deadline);
            assertEquals(0, backupConsumer.currentLag(checkpointTopicPartition).orElse(1), "Unable to read all checkpoints within " + CHECKPOINT_DURATION_MS + "ms");
        }
    }
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