Added another test about removal of headers based on reg exp
@@ -196,6 +196,54 @@ public class CamelSinkTaskTest {
sinkTask.stop();
}
+ @Test
+ public void testBodyAndHeadersExclusionsRegex() {
+ Map<String, String> props = new HashMap<>();
+ props.put(TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
+ props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF, "My*");
+
+ CamelSinkTask sinkTask = new CamelSinkTask();
+ sinkTask.start(props);
+
+ Byte myByte = new Byte("100");
+ Float myFloat = new Float("100");
+ Short myShort = new Short("100");
+ Double myDouble = new Double("100");
+ int myInteger = 100;
+ Long myLong = new Long("100");
+ BigDecimal myBigDecimal = new BigDecimal(1234567890);
+ Schema schema = Decimal.schema(myBigDecimal.scale());
+
+ List<SinkRecord> records = new ArrayList<SinkRecord>();
+ SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
+ record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
+ record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
+ record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
+ record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
+ record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
+ record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
+ record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
+ record.headers().add(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBigDecimal", Decimal.fromLogical(schema, myBigDecimal), schema);
+ records.add(record);
+ sinkTask.put(records);
+
+ ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
+ Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
+ assertEquals("camel", exchange.getMessage().getBody());
+ assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
+ assertNull(exchange.getIn().getHeader("MyBoolean", Boolean.class));
+ assertNull(exchange.getIn().getHeader("MyByte", Byte.class));
+ assertNull(exchange.getIn().getHeader("MyFloat", Float.class));
+ assertNull(exchange.getIn().getHeader("MyShort", Short.class));
+ assertNull(exchange.getIn().getHeader("MyDouble", Double.class));
+ assertNull(exchange.getIn().getHeader("MyInteger"));
+ assertNull(exchange.getIn().getHeader("MyLong", Long.class));
+ assertNull(exchange.getIn().getHeader("MyBigDecimal", BigDecimal.class));
+
+ sinkTask.stop();
+ }
+
@Test
public void testBodyAndProperties() {
Map<String, String> props = new HashMap<>();
