@Override
    protected Map<String, ConfigValue> validateSourceConnectorConfig(SourceConnector connector, ConfigDef configDef, Map<String, String> config) {
        Map<String, ConfigValue> result = super.validateSourceConnectorConfig(connector, configDef, config);
        validateSourceConnectorExactlyOnceSupport(config, result, connector);
        validateSourceConnectorTransactionBoundary(config, result, connector);
        return result;
    }
@Test
    public void testExactlyOnceSourceSupportValidationHandlesConnectorErrorsGracefully() {
        herder = exactlyOnceHerder();
        Map<String, String> config = new HashMap<>();
        config.put(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG, REQUIRED.toString());

        SourceConnector connectorMock = mock(SourceConnector.class);
        String errorMessage = "time to add a new unit test :)";
        when(connectorMock.exactlyOnceSupport(eq(config))).thenThrow(new NullPointerException(errorMessage));

        Map<String, ConfigValue> validatedConfigs = herder.validateSourceConnectorConfig(
                connectorMock, SourceConnectorConfig.configDef(), config);

        List<String> errors = validatedConfigs.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG).errorMessages();
        assertFalse(errors.isEmpty());
        assertTrue(
            errors.get(0).contains(errorMessage),
            "Error message did not contain expected text: " + errors.get(0));
        assertEquals(1, errors.size());
    }