EventRouter implements Transformation<R> { @Override public void configure(Map<String, ?> configMap) { final Configuration config = Configuration.from(configMap); smtManager = new SmtManager<>(config); io.debezium.config.Field.Set allFields = io.debezium.config.Field.setOf(EventRouterConfigDefinition.CONFIG_FIELDS); if (!config.validateAndRecord(allFields, LOGGER::error)) { throw new ConnectException("Unable to validate config."); } invalidOperationBehavior = EventRouterConfigDefinition.InvalidOperationBehavior.parse( config.getString(EventRouterConfigDefinition.OPERATION_INVALID_BEHAVIOR)); fieldEventId = config.getString(EventRouterConfigDefinition.FIELD_EVENT_ID); fieldEventKey = config.getString(EventRouterConfigDefinition.FIELD_EVENT_KEY); fieldEventTimestamp = config.getString(EventRouterConfigDefinition.FIELD_EVENT_TIMESTAMP); fieldPayload = config.getString(EventRouterConfigDefinition.FIELD_PAYLOAD); fieldPayloadId = config.getString(EventRouterConfigDefinition.FIELD_PAYLOAD_ID); fieldSchemaVersion = config.getString(EventRouterConfigDefinition.FIELD_SCHEMA_VERSION); routeByField = config.getString(EventRouterConfigDefinition.ROUTE_BY_FIELD); routeTombstoneOnEmptyPayload = config.getBoolean(EventRouterConfigDefinition.ROUTE_TOMBSTONE_ON_EMPTY_PAYLOAD); final Map<String, String> regexRouterConfig = new HashMap<>(); regexRouterConfig.put("regex", config.getString(EventRouterConfigDefinition.ROUTE_TOPIC_REGEX)); regexRouterConfig.put("replacement", config.getString(EventRouterConfigDefinition.ROUTE_TOPIC_REPLACEMENT)); regexRouter.configure(regexRouterConfig); final Map<String, String> afterExtractorConfig = new HashMap<>(); afterExtractorConfig.put("field", Envelope.FieldName.AFTER); afterExtractor.configure(afterExtractorConfig); additionalFields = parseAdditionalFieldsConfig(config); onlyHeadersInOutputMessage = !additionalFields.stream().anyMatch(field -> field.getPlacement() == EventRouterConfigDefinition.AdditionalFieldPlacement.ENVELOPE); }  @Override R apply(R r); @Override ConfigDef config(); @Override void close(); @Override void configure(Map<String, ?> configMap);  }
@Test(expected = ConnectException.class) public void shouldFailOnInvalidConfigurationForTopicRegex() { final EventRouter<SourceRecord> router = new EventRouter<>(); final Map<String, String> config = new HashMap<>(); config.put(EventRouterConfigDefinition.ROUTE_TOPIC_REGEX.name(), " [[a-z]"); router.configure(config); }
@Test(expected = ConnectException.class) public void shouldFailOnInvalidConfigurationForAdditionalFields() { final EventRouter<SourceRecord> router = new EventRouter<>(); final Map<String, String> config = new HashMap<>(); config.put(EventRouterConfigDefinition.FIELDS_ADDITIONAL_PLACEMENT.name(), "type"); router.configure(config); }
@Test(expected = ConnectException.class) public void shouldFailOnInvalidConfigurationForAdditionalFieldsEmpty() { final EventRouter<SourceRecord> router = new EventRouter<>(); final Map<String, String> config = new HashMap<>(); config.put(EventRouterConfigDefinition.FIELDS_ADDITIONAL_PLACEMENT.name(), ""); router.configure(config); }
@Test(expected = ConnectException.class) public void shouldFailOnInvalidConfigurationForOperationBehavior() { final EventRouter<SourceRecord> router = new EventRouter<>(); final Map<String, String> config = new HashMap<>(); config.put(EventRouterConfigDefinition.OPERATION_INVALID_BEHAVIOR.name(), "invalidOption"); router.configure(config); }