@Test
    public void testReconfigureConnectorWithFailingTaskConfigs(@TempDir Path tmp) throws Exception {
        final int offsetCommitIntervalMs = 100;
        workerProps.put(CONFIG_PROVIDERS_CONFIG, "file");
        workerProps.put(CONFIG_PROVIDERS_CONFIG + ".file.class", FileConfigProvider.class.getName());
        workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, Integer.toString(offsetCommitIntervalMs));

        final int numWorkers = 1;
        connect = connectBuilder
                .numWorkers(numWorkers)
                .build();
        // start the clusters
        connect.start();

        final String firstConnectorTopic = "connector-topic-1";
        connect.kafka().createTopic(firstConnectorTopic);

        final File secretsFile = tmp.resolve("test-secrets").toFile();
        final Properties secrets = new Properties();
        final String throughputSecretKey = "secret-throughput";
        secrets.put(throughputSecretKey, "10");
        try (FileOutputStream secretsOutputStream = new FileOutputStream(secretsFile)) {
            secrets.store(secretsOutputStream, null);
        }

        ConnectorHandle connectorHandle = RuntimeHandles.get().connectorHandle(CONNECTOR_NAME);
        connectorHandle.expectedCommits(NUM_TASKS * 2);

        Map<String, String> connectorConfig = defaultSourceConnectorProps(firstConnectorTopic);
        connectorConfig.put(
                "throughput",
                "${file:" + secretsFile.getAbsolutePath() + ":" + throughputSecretKey + "}"
        );
        connect.configureConnector(CONNECTOR_NAME, connectorConfig);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                CONNECTOR_NAME,
                NUM_TASKS,
                "Connector or its tasks did not start in time"
        );
        connectorHandle.awaitCommits(RECORD_TRANSFER_TIMEOUT_MS);

        // Delete the secrets file, which should render the old task configs invalid
        assertTrue(secretsFile.delete(), "Failed to delete secrets file");

        // Use a start latch here instead of assertConnectorAndExactlyNumTasksAreRunning
        // since failure to reconfigure the tasks (which may occur if the bug this test was written
        // to help catch resurfaces) will not cause existing tasks to fail or stop running
        StartAndStopLatch restarts = connectorHandle.expectedStarts(1);

        final String secondConnectorTopic = "connector-topic-2";
        connect.kafka().createTopic(secondConnectorTopic, 1);

        // Stop using the config provider for this connector, and instruct it to start writing to the
        // old topic again
        connectorConfig.put("throughput", "10");
        connectorConfig.put(TOPIC_CONFIG, secondConnectorTopic);
        connect.configureConnector(CONNECTOR_NAME, connectorConfig);
        assertTrue(
                restarts.await(10, TimeUnit.SECONDS),
                "Connector tasks were not restarted in time"
        );

        // Wait for at least one task to commit offsets after being restarted
        connectorHandle.expectedCommits(1);
        connectorHandle.awaitCommits(RECORD_TRANSFER_TIMEOUT_MS);

        final long endOffset = connect.kafka().endOffset(new TopicPartition(secondConnectorTopic, 0));
        assertTrue(
                endOffset > 0,
                "Source connector should have published at least one record to new Kafka topic "
                        + "after being reconfigured"
        );
    }
If a connector has existing tasks, and then generates new task configs, workers compare the
new and existing configs before publishing them to the config topic. If there is no difference,
workers do not publish task configs (this is a workaround to prevent infinite loops with eager
rebalancing).
<p>
This test tries to guarantee that, if the old task configs become invalid because of
an invalid config provider reference, it will still be possible to reconfigure the connector.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java