@Test
    public void testRuntimePropertyReconfiguration() throws Exception {
        final int offsetCommitIntervalMs = 1_000;
        // force fast offset commits
        workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, Integer.toString(offsetCommitIntervalMs));
        connect = connectBuilder.build();
        // start the clusters
        connect.start();

        final String topic = "kafka9228";
        connect.kafka().createTopic(topic, 1);
        connect.kafka().produce(topic, "non-json-value");

        Map<String, String> connectorConfig = new HashMap<>();
        connectorConfig.put(CONNECTOR_CLASS_CONFIG, EmptyTaskConfigsConnector.class.getName());
        connectorConfig.put(TASKS_MAX_CONFIG, "1");
        connectorConfig.put(TOPICS_CONFIG, topic);
        // Initially configure the connector to use the JSON converter, which should cause task failure(s)
        connectorConfig.put(VALUE_CONVERTER_CLASS_CONFIG, JsonConverter.class.getName());
        connectorConfig.put(
                VALUE_CONVERTER_CLASS_CONFIG + "." + JsonConverterConfig.SCHEMAS_ENABLE_CONFIG,
                "false"
        );

        connect.configureConnector(CONNECTOR_NAME, connectorConfig);
        connect.assertions().assertConnectorIsRunningAndTasksHaveFailed(
                CONNECTOR_NAME,
                1,
                "Connector did not start or task did not fail in time"
        );
        assertEquals(
                new ConnectorOffsets(Collections.emptyList()),
                connect.connectorOffsets(CONNECTOR_NAME),
                "Connector should not have any committed offsets when only task fails on first record"
        );

        // Reconfigure the connector to use the string converter, which should not cause any more task failures
        connectorConfig.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
        connectorConfig.remove(
                KEY_CONVERTER_CLASS_CONFIG + "." + JsonConverterConfig.SCHEMAS_ENABLE_CONFIG
        );
        connect.configureConnector(CONNECTOR_NAME, connectorConfig);
        connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                CONNECTOR_NAME,
                1,
                "Connector or tasks did not start in time"
        );

        Map<String, Object> expectedOffsetKey = new HashMap<>();
        expectedOffsetKey.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
        expectedOffsetKey.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
        Map<String, Object> expectedOffsetValue = Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 1);
        ConnectorOffset expectedOffset = new ConnectorOffset(expectedOffsetKey, expectedOffsetValue);
        ConnectorOffsets expectedOffsets = new ConnectorOffsets(Collections.singletonList(expectedOffset));

        // Wait for it to commit offsets, signaling that it has successfully processed the record we produced earlier
        waitForCondition(
                () -> expectedOffsets.equals(connect.connectorOffsets(CONNECTOR_NAME)),
                offsetCommitIntervalMs * 2,
                "Task did not successfully process record and/or commit offsets in time"
        );
    }
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