@Test
    public void testPreflightValidation() {
        connectBuilder.numWorkers(1);
        startConnect();

        Map<String, String> props = new HashMap<>();
        props.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getName());
        props.put(TASKS_MAX_CONFIG, "1");
        props.put(TOPIC_CONFIG, "topic");
        props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
        props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
        props.put(NAME_CONFIG, CONNECTOR_NAME);

        // Test out the "exactly.once.support" property
        props.put(EXACTLY_ONCE_SUPPORT_CONFIG, "required");

        // Connector will return null from SourceConnector::exactlyOnceSupport
        props.put(CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG, MonitorableSourceConnector.EXACTLY_ONCE_NULL);
        ConfigInfos validation = connect.validateConnectorConfig(MonitorableSourceConnector.class.getSimpleName(), props);
        assertEquals(1, validation.errorCount(),
                "Preflight validation should have exactly one error");
        ConfigInfo propertyValidation = findConfigInfo(EXACTLY_ONCE_SUPPORT_CONFIG, validation);
        assertFalse(propertyValidation.configValue().errors().isEmpty(),
                "Preflight validation for exactly-once support property should have at least one error message");

        // Connector will return UNSUPPORTED from SourceConnector::exactlyOnceSupport
        props.put(CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG, MonitorableSourceConnector.EXACTLY_ONCE_UNSUPPORTED);
        validation = connect.validateConnectorConfig(MonitorableSourceConnector.class.getSimpleName(), props);
        assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
        propertyValidation = findConfigInfo(EXACTLY_ONCE_SUPPORT_CONFIG, validation);
        assertFalse(propertyValidation.configValue().errors().isEmpty(), 
                "Preflight validation for exactly-once support property should have at least one error message");

        // Connector will throw an exception from SourceConnector::exactlyOnceSupport
        props.put(CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG, MonitorableSourceConnector.EXACTLY_ONCE_FAIL);
        validation = connect.validateConnectorConfig(MonitorableSourceConnector.class.getSimpleName(), props);
        assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
        propertyValidation = findConfigInfo(EXACTLY_ONCE_SUPPORT_CONFIG, validation);
        assertFalse(propertyValidation.configValue().errors().isEmpty(),
                "Preflight validation for exactly-once support property should have at least one error message");

        // Connector will return SUPPORTED from SourceConnector::exactlyOnceSupport
        props.put(CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG, MonitorableSourceConnector.EXACTLY_ONCE_SUPPORTED);
        validation = connect.validateConnectorConfig(MonitorableSourceConnector.class.getSimpleName(), props);
        assertEquals(0, validation.errorCount(), "Preflight validation should have zero errors");

        // Test out the transaction boundary definition property
        props.put(TRANSACTION_BOUNDARY_CONFIG, CONNECTOR.toString());

        // Connector will return null from SourceConnector::canDefineTransactionBoundaries
        props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, MonitorableSourceConnector.TRANSACTION_BOUNDARIES_NULL);
        validation = connect.validateConnectorConfig(MonitorableSourceConnector.class.getSimpleName(), props);
        assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
        propertyValidation = findConfigInfo(TRANSACTION_BOUNDARY_CONFIG, validation);
        assertFalse(propertyValidation.configValue().errors().isEmpty(),
                "Preflight validation for transaction boundary property should have at least one error message");

        // Connector will return UNSUPPORTED from SourceConnector::canDefineTransactionBoundaries
        props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, MonitorableSourceConnector.TRANSACTION_BOUNDARIES_UNSUPPORTED);
        validation = connect.validateConnectorConfig(MonitorableSourceConnector.class.getSimpleName(), props);
        assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
        propertyValidation = findConfigInfo(TRANSACTION_BOUNDARY_CONFIG, validation);
        assertFalse(propertyValidation.configValue().errors().isEmpty(),
                "Preflight validation for transaction boundary property should have at least one error message");

        // Connector will throw an exception from SourceConnector::canDefineTransactionBoundaries
        props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, MonitorableSourceConnector.TRANSACTION_BOUNDARIES_FAIL);
        validation = connect.validateConnectorConfig(MonitorableSourceConnector.class.getSimpleName(), props);
        assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
        propertyValidation = findConfigInfo(TRANSACTION_BOUNDARY_CONFIG, validation);
        assertFalse(propertyValidation.configValue().errors().isEmpty(), 
                "Preflight validation for transaction boundary property should have at least one error message");

        // Connector will return SUPPORTED from SourceConnector::canDefineTransactionBoundaries
        props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, MonitorableSourceConnector.TRANSACTION_BOUNDARIES_SUPPORTED);
        validation = connect.validateConnectorConfig(MonitorableSourceConnector.class.getSimpleName(), props);
        assertEquals(0, validation.errorCount(), "Preflight validation should have zero errors");
    }
A simple test for the pre-flight validation API for connectors to provide their own guarantees for exactly-once semantics.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java