@Override
    protected Map<String, ConfigValue> validateSinkConnectorConfig(SinkConnector connector, ConfigDef configDef, Map<String, String> config) {
        Map<String, ConfigValue> result = super.validateSinkConnectorConfig(connector, configDef, config);
        validateSinkConnectorGroupId(config, result);
        return result;
    }
@Test
    public void testConnectorGroupIdConflictsWithWorkerGroupId() {
        String overriddenGroupId = CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + GROUP_ID_CONFIG;
        Map<String, String> config = new HashMap<>(CONN2_CONFIG);
        config.put(overriddenGroupId, "connect-test-group");

        SinkConnector connectorMock = mock(SinkConnector.class);

        // CONN2 creation should fail because the worker group id (connect-test-group) conflicts with
        // the consumer group id we would use for this sink
        Map<String, ConfigValue> validatedConfigs = herder.validateSinkConnectorConfig(
                connectorMock, SinkConnectorConfig.configDef(), config);

        ConfigValue overriddenGroupIdConfig = validatedConfigs.get(overriddenGroupId);
        assertEquals(
                Collections.singletonList("Consumer group connect-test-group conflicts with Connect worker group connect-test-group"),
                overriddenGroupIdConfig.errorMessages());

        ConfigValue nameConfig = validatedConfigs.get(ConnectorConfig.NAME_CONFIG);
        assertEquals(
                Collections.emptyList(),
                nameConfig.errorMessages()
        );
    }