@Test
    public void testPauseStopResume() throws Exception {
        connect = connectBuilder.build();
        // start the clusters
        connect.start();

        // Want to make sure to use multiple tasks
        final int numTasks = 4;
        Map<String, String> props = defaultSourceConnectorProps(TOPIC_NAME);
        props.put(TASKS_MAX_CONFIG, Integer.toString(numTasks));

        // Start with RUNNING
        connect.configureConnector(CONNECTOR_NAME, props);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                CONNECTOR_NAME,
                numTasks,
                "Connector tasks did not start in time"
        );

        // Transition to STOPPED
        connect.stopConnector(CONNECTOR_NAME);
        // Issue a second request to ensure that this operation is idempotent
        connect.stopConnector(CONNECTOR_NAME);
        connect.assertions().assertConnectorIsStopped(
                CONNECTOR_NAME,
                "Connector did not stop in time"
        );
        // If the connector is truly stopped, we should also see an empty set of tasks and task configs
        assertEquals(Collections.emptyList(), connect.connectorInfo(CONNECTOR_NAME).tasks());
        assertEquals(Collections.emptyList(), connect.taskConfigs(CONNECTOR_NAME));

        // Transition to RUNNING
        connect.resumeConnector(CONNECTOR_NAME);
        // Issue a second request to ensure that this operation is idempotent
        connect.resumeConnector(CONNECTOR_NAME);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                CONNECTOR_NAME,
                numTasks,
                "Connector tasks did not resume in time"
        );

        // Transition to PAUSED
        connect.pauseConnector(CONNECTOR_NAME);
        // Issue a second request to ensure that this operation is idempotent
        connect.pauseConnector(CONNECTOR_NAME);
        connect.assertions().assertConnectorAndExactlyNumTasksArePaused(
                CONNECTOR_NAME,
                numTasks,
                "Connector did not pause in time"
        );

        // Transition to STOPPED
        connect.stopConnector(CONNECTOR_NAME);
        connect.assertions().assertConnectorIsStopped(
                CONNECTOR_NAME,
                "Connector did not stop in time"
        );
        assertEquals(Collections.emptyList(), connect.connectorInfo(CONNECTOR_NAME).tasks());
        assertEquals(Collections.emptyList(), connect.taskConfigs(CONNECTOR_NAME));

        // Transition to PAUSED
        connect.pauseConnector(CONNECTOR_NAME);
        connect.assertions().assertConnectorAndExactlyNumTasksArePaused(
                CONNECTOR_NAME,
                0,
                "Connector did not pause in time"
        );

        // Transition to RUNNING
        connect.resumeConnector(CONNECTOR_NAME);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                CONNECTOR_NAME,
                numTasks,
                "Connector tasks did not resume in time"
        );

        // Delete the connector
        connect.deleteConnector(CONNECTOR_NAME);
        connect.assertions().assertConnectorDoesNotExist(
                CONNECTOR_NAME,
                "Connector wasn't deleted in time"
        );
    }
Verify that the target state (started, paused, stopped) of a connector can be updated, with
an emphasis on ensuring that the transitions between each state are correct.
<p>
The transitions we need to cover are:
<ol>
    <li>RUNNING -> PAUSED</li>
    <li>RUNNING -> STOPPED</li>
    <li>PAUSED -> RUNNING</li>
    <li>PAUSED -> STOPPED</li>
    <li>STOPPED -> RUNNING</li>
    <li>STOPPED -> PAUSED</li>
</ol>
With some reordering, we can perform each transition just once:
<ul>
    <li>Start with RUNNING</li>
    <li>Transition to STOPPED (2)</li>
    <li>Transition to RUNNING (5)</li>
    <li>Transition to PAUSED (1)</li>
    <li>Transition to STOPPED (4)</li>
    <li>Transition to PAUSED (6)</li>
    <li>Transition to RUNNING (3)</li>
</ul>
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java