private void resetAndVerifySinkConnectorOffsets(Map<String, String> connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
        int numPartitions = 3;
        kafkaCluster.createTopic(topic, numPartitions);

        // Produce records to each partition
        for (int partition = 0; partition < numPartitions; 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, numPartitions, NUM_RECORDS_PER_PARTITION,
                "Sink connector consumer group offsets should catch up to the topic end offsets");

        connect.stopConnector(connectorName);
        connect.assertions().assertConnectorIsStopped(
                connectorName,
                "Connector did not stop in time"
        );

        // Reset the sink connector's offsets, with retry logic (since we just stopped the connector)
        String response = modifySinkConnectorOffsetsWithRetry(null);
        assertThat(response, containsString("The Connect framework-managed offsets for this connector have been reset successfully. " +
                "However, if this connector manages offsets externally, they will need to be manually reset in the system that the connector uses."));

        verifyEmptyConnectorOffsets(connectorName);

        // Reset the sink connector's offsets again while it is still in a STOPPED state and ensure that there is no error
        response = connect.resetConnectorOffsets(connectorName);
        assertThat(response, containsString("The Connect framework-managed offsets for this connector have been reset successfully. " +
                "However, if this connector manages offsets externally, they will need to be manually reset in the system that the connector uses."));

        verifyEmptyConnectorOffsets(connectorName);

        // Resume the connector and expect its offsets to catch up to the latest offsets
        connect.resumeConnector(connectorName);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                connectorName,
                NUM_TASKS,
                "Connector tasks did not resume in time"
        );
        verifyExpectedSinkConnectorOffsets(connectorName, topic, numPartitions, 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