ByLogicalTableRouter implements Transformation<R> { @Override public void configure(Map<String, ?> props) { Configuration config = Configuration.from(props); final Field.Set configFields = Field.setOf( TOPIC_REGEX, TOPIC_REPLACEMENT, KEY_ENFORCE_UNIQUENESS, KEY_FIELD_REGEX, KEY_FIELD_REPLACEMENT); if (!config.validateAndRecord(configFields, LOGGER::error)) { throw new ConnectException("Unable to validate config."); } topicRegex = Pattern.compile(config.getString(TOPIC_REGEX)); topicReplacement = config.getString(TOPIC_REPLACEMENT); String keyFieldRegexString = config.getString(KEY_FIELD_REGEX); if (keyFieldRegexString != null) { keyFieldRegexString = keyFieldRegexString.trim(); } if (keyFieldRegexString != null && !keyFieldRegexString.isEmpty()) { keyFieldRegex = Pattern.compile(config.getString(KEY_FIELD_REGEX)); keyFieldReplacement = config.getString(KEY_FIELD_REPLACEMENT); } keyFieldName = config.getString(KEY_FIELD_NAME); keyEnforceUniqueness = config.getBoolean(KEY_ENFORCE_UNIQUENESS); smtManager = new SmtManager<>(config); }  @Override void configure(Map<String, ?> props); @Override R apply(R record); @Override void close(); @Override ConfigDef config();  }
@Test(expected = ConnectException.class) public void testBrokenKeyReplacementConfigurationNullValue() { final ByLogicalTableRouter<SourceRecord> subject = new ByLogicalTableRouter<>(); final Map<String, String> props = new HashMap<>(); props.put("topic.regex", "someValidRegex(.*)"); props.put("topic.replacement", "$1"); props.put("key.field.regex", "If this is set, key.field.replacement must be non-empty"); subject.configure(props); }
@Test(expected = ConnectException.class) public void testBrokenKeyReplacementConfigurationEmptyValue() { final ByLogicalTableRouter<SourceRecord> subject = new ByLogicalTableRouter<>(); final Map<String, String> props = new HashMap<>(); props.put("topic.regex", "someValidRegex(.*)"); props.put("topic.replacement", "$1"); props.put("key.field.regex", "If this is set, key.field.replacement must be non-empty"); props.put("key.field.replacement", ""); subject.configure(props); }
@Test(expected = ConnectException.class) public void testBrokenTopicReplacementConfigurationNullValue() { final ByLogicalTableRouter<SourceRecord> subject = new ByLogicalTableRouter<>(); final Map<String, String> props = new HashMap<>(); props.put("topic.regex", "someValidRegex(.*)"); subject.configure(props); }
@Test(expected = ConnectException.class) public void testBrokenTopicReplacementConfigurationEmptyValue() { final ByLogicalTableRouter<SourceRecord> subject = new ByLogicalTableRouter<>(); final Map<String, String> props = new HashMap<>(); props.put("topic.regex", "someValidRegex(.*)"); props.put("topic.replacement", ""); subject.configure(props); }