Topics.regex: Added unit test
@@ -68,6 +68,35 @@ public class CamelSinkTaskTest {
sinkTask.stop();
}
+ @Test
+ public void testTopicsRegex() {
+ Map<String, String> props = new HashMap<>();
+ props.put("topics.regex", "topic1*");
+ props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
+
+ CamelSinkTask sinkTask = new CamelSinkTask();
+ sinkTask.start(props);
+
+ List<SinkRecord> records = new ArrayList<SinkRecord>();
+ SinkRecord record = new SinkRecord("topic1", 1, null, "test", null, "camel", 42);
+ SinkRecord record1 = new SinkRecord("topic12", 1, null, "test", null, "cameltopicregex", 42);
+ records.add(record);
+ records.add(record1);
+ sinkTask.put(records);
+
+ ConsumerTemplate consumer = sinkTask.getCms().createConsumerTemplate();
+ Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
+ assertEquals("camel", exchange.getMessage().getBody());
+ assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
+ assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
+ .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
+ Exchange exchange1 = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
+ assertEquals("cameltopicregex", exchange1.getMessage().getBody());
+ assertEquals("test", exchange1.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
+
+ sinkTask.stop();
+ }
+
@Test
public void testBodyAndHeaders() {
Map<String, String> props = new HashMap<>();
