@Test
    public void testSourceTaskNotBlockedOnShutdownWithNonExistentTopic() throws Exception {
        // When automatic topic creation is disabled on the broker
        brokerProps.put("auto.create.topics.enable", "false");
        connect = connectBuilder
            .brokerProps(brokerProps)
            .numWorkers(1)
            .numBrokers(1)
            .build();
        connect.start();

        // and when the connector is not configured to create topics
        Map<String, String> props = defaultSourceConnectorProps("nonexistenttopic");
        props.remove(DEFAULT_TOPIC_CREATION_PREFIX + REPLICATION_FACTOR_CONFIG);
        props.remove(DEFAULT_TOPIC_CREATION_PREFIX + PARTITIONS_CONFIG);
        props.put("throughput", "-1");

        ConnectorHandle connector = RuntimeHandles.get().connectorHandle(CONNECTOR_NAME);
        connector.expectedRecords(NUM_TASKS * MESSAGES_PER_POLL);
        connect.configureConnector(CONNECTOR_NAME, props);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(CONNECTOR_NAME,
            NUM_TASKS, "Connector tasks did not start in time");
        connector.awaitRecords(TimeUnit.MINUTES.toMillis(1));

        // Then if we delete the connector, it and each of its tasks should be stopped by the framework
        // even though the producer is blocked because there is no topic
        StartAndStopLatch stopCounter = connector.expectedStops(1);
        connect.deleteConnector(CONNECTOR_NAME);

        assertTrue(stopCounter.await(1, TimeUnit.MINUTES), "Connector and all tasks were not stopped in time");
    }
Verify that the number of tasks listed in the REST API is updated correctly after changes to
the "tasks.max" connector configuration.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java