Added support for headers in source connector
@@ -29,8 +29,13 @@ import org.apache.kafka.connect.source.SourceTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.math.BigDecimal;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Date;
import java.util.List;
import java.util.Map;
@@ -96,12 +101,14 @@ public class CamelSourceTask extends SourceTask {
log.info("\t exchange id: {}", exchange.getExchangeId());
log.info("\t message id: {}", exchange.getMessage().getMessageId());
log.info("\t message body: {}", exchange.getMessage().getBody());
+ log.info("\t message headers: {}", exchange.getMessage().getHeaders());
//TODO: see if there is a better way to use sourcePartition and sourceOffset
Map<String, String> sourcePartition = Collections.singletonMap("filename", exchange.getFromEndpoint().toString());
Map<String, String> sourceOffset = Collections.singletonMap("position", exchange.getExchangeId());
SourceRecord record = new SourceRecord(sourcePartition, sourceOffset, topic, Schema.BYTES_SCHEMA, exchange.getMessage().getBody());
+ setHeaders(record, exchange);
records.add(record);
return records;
@@ -120,5 +127,58 @@ public class CamelSourceTask extends SourceTask {
throw new ConnectException("Failed to stop Camel context", e);
}
}
+
+ private void setHeaders(SourceRecord record, Exchange exchange) {
+ if (!exchange.getMessage().hasHeaders()) {
+ return;
+ }
+
+ for (Map.Entry<String, Object> entry : exchange.getMessage().getHeaders().entrySet()) {
+ String key = entry.getKey();
+ Object value = entry.getValue();
+
+ if (value instanceof String) {
+ record.headers().addString(key, (String) value);
+ }
+ if (value instanceof Boolean) {
+ record.headers().addBoolean(key, (boolean) value);
+ }
+ if (value instanceof Byte) {
+ record.headers().addByte(key, (byte) value);
+ }
+ if (value instanceof Byte[]) {
+ record.headers().addBytes(key, (byte[]) value);
+ }
+ if (value instanceof Date) {
+ SimpleDateFormat sdf = new SimpleDateFormat("YYYY-MM-dd");
+ String convertedDate = sdf.format(value);
+ record.headers().addString(key, (String) convertedDate);
+ }
+ if (value instanceof BigDecimal) {
+ record.headers().addDecimal(key, (BigDecimal) value);
+ }
+ if (value instanceof Double) {
+ record.headers().addDouble(key, (double) value);
+ }
+ if (value instanceof Float) {
+ record.headers().addFloat(key, (float) value);
+ }
+ if (value instanceof Integer) {
+ record.headers().addInt(key, (int) value);
+ }
+ if (value instanceof Long) {
+ record.headers().addLong(key, (long) value);
+ }
+ if (value instanceof Short) {
+ record.headers().addShort(key, (short) value);
+ }
+ if (value instanceof Time) {
+ record.headers().addTime(key, (Time) value);
+ }
+ if (value instanceof Timestamp) {
+ record.headers().addTimestamp(key, (Timestamp) value);
+ }
+ }
+ }
}
