@Test() public void testConfigValidationMissingName() { AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class); replayAll(); Map<String, String> config = Collections.singletonMap(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSourceConnector.class.getName()); ConfigInfos result = herder.validateConnectorConfig(config); assertEquals(TestSourceConnector.class.getName(), result.name()); assertEquals(Arrays.asList(ConnectorConfig.COMMON_GROUP, ConnectorConfig.TRANSFORMS_GROUP), result.groups()); assertEquals(2, result.errorCount()); assertEquals(8, result.values().size()); assertEquals(ConnectorConfig.NAME_CONFIG, result.values().get(0).configValue().name()); assertEquals(1, result.values().get(0).configValue().errors().size()); assertEquals("required", result.values().get(6).configValue().name()); assertEquals(1, result.values().get(6).configValue().errors().size()); verifyAll(); }
@Override public ConfigInfos validateConnectorConfig(Map<String, String> connectorProps) { String connType = connectorProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG); if (connType == null) throw new BadRequestException("Connector config " + connectorProps + " contains no connector type"); List<ConfigValue> configValues = new ArrayList<>(); Map<String, ConfigKey> configKeys = new LinkedHashMap<>(); Set<String> allGroups = new LinkedHashSet<>(); Connector connector = getConnector(connType); ClassLoader savedLoader = plugins().compareAndSwapLoaders(connector); try { ConfigDef baseConfigDef = (connector instanceof SourceConnector) ? SourceConnectorConfig.configDef() : SinkConnectorConfig.configDef(); ConfigDef enrichedConfigDef = ConnectorConfig.enrich(plugins(), baseConfigDef, connectorProps, false); Map<String, ConfigValue> validatedConnectorConfig = validateBasicConnectorConfig( connector, enrichedConfigDef, connectorProps ); configValues.addAll(validatedConnectorConfig.values()); configKeys.putAll(enrichedConfigDef.configKeys()); allGroups.addAll(enrichedConfigDef.groups()); Config config = connector.validate(connectorProps); ConfigDef configDef = connector.config(); configKeys.putAll(configDef.configKeys()); allGroups.addAll(configDef.groups()); configValues.addAll(config.configValues()); return generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups)); } finally { Plugins.compareAndSwapLoaders(savedLoader); } }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConfigInfos validateConnectorConfig(Map<String, String> connectorProps) { String connType = connectorProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG); if (connType == null) throw new BadRequestException("Connector config " + connectorProps + " contains no connector type"); List<ConfigValue> configValues = new ArrayList<>(); Map<String, ConfigKey> configKeys = new LinkedHashMap<>(); Set<String> allGroups = new LinkedHashSet<>(); Connector connector = getConnector(connType); ClassLoader savedLoader = plugins().compareAndSwapLoaders(connector); try { ConfigDef baseConfigDef = (connector instanceof SourceConnector) ? SourceConnectorConfig.configDef() : SinkConnectorConfig.configDef(); ConfigDef enrichedConfigDef = ConnectorConfig.enrich(plugins(), baseConfigDef, connectorProps, false); Map<String, ConfigValue> validatedConnectorConfig = validateBasicConnectorConfig( connector, enrichedConfigDef, connectorProps ); configValues.addAll(validatedConnectorConfig.values()); configKeys.putAll(enrichedConfigDef.configKeys()); allGroups.addAll(enrichedConfigDef.groups()); Config config = connector.validate(connectorProps); ConfigDef configDef = connector.config(); configKeys.putAll(configDef.configKeys()); allGroups.addAll(configDef.groups()); configValues.addAll(config.configValues()); return generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups)); } finally { Plugins.compareAndSwapLoaders(savedLoader); } } }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConfigInfos validateConnectorConfig(Map<String, String> connectorProps) { String connType = connectorProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG); if (connType == null) throw new BadRequestException("Connector config " + connectorProps + " contains no connector type"); List<ConfigValue> configValues = new ArrayList<>(); Map<String, ConfigKey> configKeys = new LinkedHashMap<>(); Set<String> allGroups = new LinkedHashSet<>(); Connector connector = getConnector(connType); ClassLoader savedLoader = plugins().compareAndSwapLoaders(connector); try { ConfigDef baseConfigDef = (connector instanceof SourceConnector) ? SourceConnectorConfig.configDef() : SinkConnectorConfig.configDef(); ConfigDef enrichedConfigDef = ConnectorConfig.enrich(plugins(), baseConfigDef, connectorProps, false); Map<String, ConfigValue> validatedConnectorConfig = validateBasicConnectorConfig( connector, enrichedConfigDef, connectorProps ); configValues.addAll(validatedConnectorConfig.values()); configKeys.putAll(enrichedConfigDef.configKeys()); allGroups.addAll(enrichedConfigDef.groups()); Config config = connector.validate(connectorProps); ConfigDef configDef = connector.config(); configKeys.putAll(configDef.configKeys()); allGroups.addAll(configDef.groups()); configValues.addAll(config.configValues()); return generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups)); } finally { Plugins.compareAndSwapLoaders(savedLoader); } } AbstractHerder(Worker worker,
                          String workerId,
                          StatusBackingStore statusBackingStore,
                          ConfigBackingStore configBackingStore); }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConfigInfos validateConnectorConfig(Map<String, String> connectorProps) { String connType = connectorProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG); if (connType == null) throw new BadRequestException("Connector config " + connectorProps + " contains no connector type"); List<ConfigValue> configValues = new ArrayList<>(); Map<String, ConfigKey> configKeys = new LinkedHashMap<>(); Set<String> allGroups = new LinkedHashSet<>(); Connector connector = getConnector(connType); ClassLoader savedLoader = plugins().compareAndSwapLoaders(connector); try { ConfigDef baseConfigDef = (connector instanceof SourceConnector) ? SourceConnectorConfig.configDef() : SinkConnectorConfig.configDef(); ConfigDef enrichedConfigDef = ConnectorConfig.enrich(plugins(), baseConfigDef, connectorProps, false); Map<String, ConfigValue> validatedConnectorConfig = validateBasicConnectorConfig( connector, enrichedConfigDef, connectorProps ); configValues.addAll(validatedConnectorConfig.values()); configKeys.putAll(enrichedConfigDef.configKeys()); allGroups.addAll(enrichedConfigDef.groups()); Config config = connector.validate(connectorProps); ConfigDef configDef = connector.config(); configKeys.putAll(configDef.configKeys()); allGroups.addAll(configDef.groups()); configValues.addAll(config.configValues()); return generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups)); } finally { Plugins.compareAndSwapLoaders(savedLoader); } } AbstractHerder(Worker worker,
                          String workerId,
                          StatusBackingStore statusBackingStore,
                          ConfigBackingStore configBackingStore); @Override void onStartup(String connector); @Override void onPause(String connector); @Override void onResume(String connector); @Override void onShutdown(String connector); @Override void onFailure(String connector, Throwable cause); @Override void onStartup(ConnectorTaskId id); @Override void onFailure(ConnectorTaskId id, Throwable cause); @Override void onShutdown(ConnectorTaskId id); @Override void onResume(ConnectorTaskId id); @Override void onPause(ConnectorTaskId id); @Override void onDeletion(String connector); @Override void pauseConnector(String connector); @Override void resumeConnector(String connector); @Override Plugins plugins(); @Override ConnectorStateInfo connectorStatus(String connName); @Override ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id); @Override ConfigInfos validateConnectorConfig(Map<String, String> connectorProps); static ConfigInfos generateResult(String connType, Map<String, ConfigKey> configKeys, List<ConfigValue> configValues, List<String> groups); }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConfigInfos validateConnectorConfig(Map<String, String> connectorProps) { String connType = connectorProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG); if (connType == null) throw new BadRequestException("Connector config " + connectorProps + " contains no connector type"); List<ConfigValue> configValues = new ArrayList<>(); Map<String, ConfigKey> configKeys = new LinkedHashMap<>(); Set<String> allGroups = new LinkedHashSet<>(); Connector connector = getConnector(connType); ClassLoader savedLoader = plugins().compareAndSwapLoaders(connector); try { ConfigDef baseConfigDef = (connector instanceof SourceConnector) ? SourceConnectorConfig.configDef() : SinkConnectorConfig.configDef(); ConfigDef enrichedConfigDef = ConnectorConfig.enrich(plugins(), baseConfigDef, connectorProps, false); Map<String, ConfigValue> validatedConnectorConfig = validateBasicConnectorConfig( connector, enrichedConfigDef, connectorProps ); configValues.addAll(validatedConnectorConfig.values()); configKeys.putAll(enrichedConfigDef.configKeys()); allGroups.addAll(enrichedConfigDef.groups()); Config config = connector.validate(connectorProps); ConfigDef configDef = connector.config(); configKeys.putAll(configDef.configKeys()); allGroups.addAll(configDef.groups()); configValues.addAll(config.configValues()); return generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups)); } finally { Plugins.compareAndSwapLoaders(savedLoader); } } AbstractHerder(Worker worker,
                          String workerId,
                          StatusBackingStore statusBackingStore,
                          ConfigBackingStore configBackingStore); @Override void onStartup(String connector); @Override void onPause(String connector); @Override void onResume(String connector); @Override void onShutdown(String connector); @Override void onFailure(String connector, Throwable cause); @Override void onStartup(ConnectorTaskId id); @Override void onFailure(ConnectorTaskId id, Throwable cause); @Override void onShutdown(ConnectorTaskId id); @Override void onResume(ConnectorTaskId id); @Override void onPause(ConnectorTaskId id); @Override void onDeletion(String connector); @Override void pauseConnector(String connector); @Override void resumeConnector(String connector); @Override Plugins plugins(); @Override ConnectorStateInfo connectorStatus(String connName); @Override ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id); @Override ConfigInfos validateConnectorConfig(Map<String, String> connectorProps); static ConfigInfos generateResult(String connType, Map<String, ConfigKey> configKeys, List<ConfigValue> configValues, List<String> groups);  }