private void warmUpConsumer(String clusterName, EmbeddedKafkaCluster kafkaCluster, Map<String, Object> consumerProps, String topic) {
        try (Consumer<?, ?> dummyConsumer = kafkaCluster.createConsumerAndSubscribeTo(consumerProps, topic)) {
            // poll to ensure we've joined the group
            dummyConsumer.poll(CONSUMER_POLL_TIMEOUT_MS);

            // force the consumer to have a known position on every topic partition
            // so that it will be able to commit offsets for that position
            // (it's possible that poll returns before that has happened)
            Set<TopicPartition> topicPartitionsPendingPosition = IntStream.range(0, NUM_PARTITIONS)
                    .mapToObj(partition -> new TopicPartition(topic, partition))
                    .collect(Collectors.toSet());
            Timer positionTimer = Time.SYSTEM.timer(60_000);
            while (!positionTimer.isExpired() && !topicPartitionsPendingPosition.isEmpty()) {
                Set<TopicPartition> topicPartitionsWithPosition = new HashSet<>();

                topicPartitionsPendingPosition.forEach(topicPartition -> {
                    try {
                        positionTimer.update();
                        dummyConsumer.position(topicPartition, Duration.ofMillis(positionTimer.remainingMs()));
                        topicPartitionsWithPosition.add(topicPartition);
                    } catch (KafkaException e) {
                        log.warn("Failed to calculate consumer position for {} on cluster {}", topicPartition, clusterName);
                    }
                });

                topicPartitionsPendingPosition.removeAll(topicPartitionsWithPosition);
            }
            assertEquals(
                    Collections.emptySet(),
                    topicPartitionsPendingPosition,
                    "Failed to calculate consumer position for one or more partitions on cluster " + clusterName + " in time"
            );

            // And finally, commit offsets
            dummyConsumer.commitSync();
        }
    }
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