public void alterAndVerifySourceConnectorOffsets(Map<String, String> connectorConfigs) throws Exception {
        // Create source connector
        connect.configureConnector(connectorName, connectorConfigs);
        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
                "Connector tasks did not start in time.");

        verifyExpectedSourceConnectorOffsets(connectorName, NUM_TASKS, NUM_RECORDS_PER_PARTITION,
                "Source connector offsets should reflect the expected number of records produced");

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

        List<ConnectorOffset> offsetsToAlter = new ArrayList<>();
        // The MonitorableSourceConnector has a source partition per task
        for (int i = 0; i < NUM_TASKS; i++) {
            offsetsToAlter.add(
                    new ConnectorOffset(Collections.singletonMap("task.id", connectorName + "-" + i),
                            Collections.singletonMap("saved", 5))
            );
        }

        String response = connect.alterConnectorOffsets(connectorName, new ConnectorOffsets(offsetsToAlter));
        assertThat(response, containsString("The Connect framework-managed offsets for this connector have been altered successfully. " +
                "However, if this connector manages offsets externally, they will need to be manually altered in the system that the connector uses."));

        verifyExpectedSourceConnectorOffsets(connectorName, NUM_TASKS, 5,
                "Source connector offsets should reflect the altered offsets");

        // Update the connector's configs; this time expect SourceConnector::alterOffsets to return true
        connectorConfigs.put(MonitorableSourceConnector.ALTER_OFFSETS_RESULT, "true");
        connect.configureConnector(connectorName, connectorConfigs);

        // Alter offsets again while connector is in stopped state
        offsetsToAlter = new ArrayList<>();
        // The MonitorableSourceConnector has a source partition per task
        for (int i = 0; i < NUM_TASKS; i++) {
            offsetsToAlter.add(
                    new ConnectorOffset(Collections.singletonMap("task.id", connectorName + "-" + i),
                            Collections.singletonMap("saved", 7))
            );
        }

        response = connect.alterConnectorOffsets(connectorName, new ConnectorOffsets(offsetsToAlter));
        assertThat(response, containsString("The offsets for this connector have been altered successfully"));

        verifyExpectedSourceConnectorOffsets(connectorName, NUM_TASKS, 7,
                "Source connector offsets should reflect the altered offsets");

        // 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"
        );
        verifyExpectedSourceConnectorOffsets(connectorName, NUM_TASKS, NUM_RECORDS_PER_PARTITION,
                "Source connector offsets should reflect the expected number of records produced");
    }
Integration tests for Kafka Connect's connector offset management REST APIs
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java