private static <T> void waitForConsumerGroupFullSync(
            EmbeddedConnectCluster connect,
            List<String> topics,
            String consumerGroupId,
            int numRecords,
            int offsetLagMax
    ) throws InterruptedException {
        int expectedRecords = numRecords * topics.size();
        Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put("isolation.level", "read_committed");
        consumerProps.put("auto.offset.reset", "earliest");
        Map<TopicPartition, Long> lastOffset = new HashMap<>();
        try (Consumer<byte[], byte[]> consumer = connect.kafka().createConsumerAndSubscribeTo(consumerProps, topics.toArray(new String[0]))) {
            final AtomicInteger totalConsumedRecords = new AtomicInteger(0);
            waitForCondition(() -> {
                ConsumerRecords<byte[], byte[]> records = consumer.poll(CONSUMER_POLL_TIMEOUT_MS);
                records.forEach(record -> lastOffset.put(new TopicPartition(record.topic(), record.partition()), record.offset()));
                return expectedRecords == totalConsumedRecords.addAndGet(records.count());
            }, RECORD_CONSUME_DURATION_MS, "Consumer cannot consume all records in time");
        }
        try (Admin adminClient = connect.kafka().createAdminClient()) {
            List<TopicPartition> tps = new ArrayList<>(NUM_PARTITIONS * topics.size());
            for (int partitionIndex = 0; partitionIndex < NUM_PARTITIONS; partitionIndex++) {
                for (String topic : topics) {
                    tps.add(new TopicPartition(topic, partitionIndex));
                }
            }

            waitForCondition(() -> {
                Map<TopicPartition, OffsetAndMetadata> consumerGroupOffsets =
                        adminClient.listConsumerGroupOffsets(consumerGroupId).partitionsToOffsetAndMetadata().get();
                Map<TopicPartition, OffsetSpec> endOffsetRequest = tps.stream()
                        .collect(Collectors.toMap(Function.identity(), ignored -> OffsetSpec.latest()));
                Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> endOffsets =
                        adminClient.listOffsets(endOffsetRequest).all().get();

                for (TopicPartition tp : tps) {
                    assertTrue(consumerGroupOffsets.containsKey(tp),
                            "TopicPartition " + tp + " does not have translated offsets");
                    long offset = consumerGroupOffsets.get(tp).offset();
                    assertTrue(offset > lastOffset.get(tp) - offsetLagMax,
                            "TopicPartition " + tp + " does not have fully-translated offsets: "
                                    + offset + " is not close enough to " + lastOffset.get(tp)
                                    + " (strictly more than " + (lastOffset.get(tp) - offsetLagMax) + ")");
                    assertTrue(offset <= endOffsets.get(tp).offset(),
                            "TopicPartition " + tp + " has downstream offsets beyond the log end, this would lead to negative lag metrics");
                }
                return true;
            }, OFFSET_SYNC_DURATION_MS, "Consumer group offset sync is not complete in time");
        }
    }
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