private void getAndVerifySinkConnectorOffsets(Map<String, String> connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
        kafkaCluster.createTopic(topic, 5);

        // Produce records to each partition
        for (int partition = 0; partition < 5; partition++) {
            for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
                kafkaCluster.produce(topic, partition, "key", "value");
            }
        }

        // Create sink connector
        connect.configureConnector(connectorName, connectorConfigs);
        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
                "Connector tasks did not start in time.");

        verifyExpectedSinkConnectorOffsets(connectorName, topic, 5, NUM_RECORDS_PER_PARTITION,
                "Sink connector consumer group offsets should catch up to the topic end offsets");

        // Produce more records to each partition
        for (int partition = 0; partition < 5; partition++) {
            for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
                kafkaCluster.produce(topic, partition, "key", "value");
            }
        }

        verifyExpectedSinkConnectorOffsets(connectorName, topic, 5, 2 * NUM_RECORDS_PER_PARTITION,
                "Sink connector consumer group offsets should catch up to the topic end offsets");
    }
Integration tests for Kafka Connect's connector offset management REST APIs
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java