@Test
    public void testResetSinkConnectorOffsetsZombieSinkTasks() throws Exception {
        connect.kafka().createTopic(topic, 1);

        // Produce records
        for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
            connect.kafka().produce(topic, 0, "key", "value");
        }

        // Configure a sink connector whose sink task blocks in its stop method
        Map<String, String> connectorConfigs = new HashMap<>();
        connectorConfigs.put(CONNECTOR_CLASS_CONFIG, BlockingConnectorTest.BlockingSinkConnector.class.getName());
        connectorConfigs.put(TOPICS_CONFIG, topic);
        connectorConfigs.put("block", "Task::stop");

        connect.configureConnector(connectorName, connectorConfigs);
        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, 1,
                "Connector tasks did not start in time.");

        // Make sure the tasks' consumers have had a chance to actually form a group
        // (otherwise, the reset request will succeed because there won't be any active consumers)
        verifyExpectedSinkConnectorOffsets(connectorName, topic, 1, NUM_RECORDS_PER_PARTITION,
                "Sink connector consumer group offsets should catch up to the topic end offsets");

        connect.stopConnector(connectorName);

        // Try to reset the offsets
        ConnectRestException e = assertThrows(ConnectRestException.class, () -> connect.resetConnectorOffsets(connectorName));
        assertThat(e.getMessage(), containsString("zombie sink task"));

        // clean up blocked threads created while testing zombie task scenarios
        BlockingConnectorTest.Block.reset();
    }
Integration tests for Kafka Connect's connector offset management REST APIs
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java