Kafka Producer doesn't handle strings properly
@@ -30,18 +30,21 @@ import org.apache.hop.pipeline.transforms.kafka.consumer.KafkaConsumerField;
import org.apache.hop.pipeline.transforms.kafka.shared.KafkaFactory;
import org.apache.kafka.clients.producer.ProducerRecord;
-
-public class KafkaProducerOutput extends BaseTransform<KafkaProducerOutputMeta, KafkaProducerOutputData>
+public class KafkaProducerOutput
+ extends BaseTransform<KafkaProducerOutputMeta, KafkaProducerOutputData>
implements ITransform<KafkaProducerOutputMeta, KafkaProducerOutputData> {
private static final Class<?> PKG = KafkaProducerOutputMeta.class; // For Translator
private KafkaFactory kafkaFactory;
- public KafkaProducerOutput( TransformMeta transformMeta,
+ public KafkaProducerOutput(
+ TransformMeta transformMeta,
KafkaProducerOutputMeta meta,
- KafkaProducerOutputData data, int copyNr,
- PipelineMeta pipelineMeta, Pipeline trans ) {
+ KafkaProducerOutputData data,
+ int copyNr,
+ PipelineMeta pipelineMeta,
+ Pipeline trans) {
super(transformMeta, meta, data, copyNr, pipelineMeta, trans);
setKafkaFactory(KafkaFactory.defaultFactory());
}
@@ -50,8 +53,8 @@ public class KafkaProducerOutput extends BaseTransform<KafkaProducerOutputMeta,
this.kafkaFactory = factory;
}
-
- @Override public boolean processRow() throws HopException {
+ @Override
+ public boolean processRow() throws HopException {
Object[] r = getRow(); // get row, set busy!
if (r == null) {
// no more input to be expected...
@@ -67,7 +70,10 @@ public class KafkaProducerOutput extends BaseTransform<KafkaProducerOutputMeta,
IValueMeta keyValueMeta = getInputRowMeta().getValueMeta(data.keyFieldIndex);
IValueMeta msgValueMeta = getInputRowMeta().getValueMeta(data.messageFieldIndex);
- data.kafkaProducer = kafkaFactory.producer( meta, this::resolve,
+ data.kafkaProducer =
+ kafkaFactory.producer(
+ meta,
+ this::resolve,
KafkaConsumerField.Type.fromValueMeta(keyValueMeta),
KafkaConsumerField.Type.fromValueMeta(msgValueMeta));
@@ -81,10 +87,15 @@ public class KafkaProducerOutput extends BaseTransform<KafkaProducerOutputMeta,
}
ProducerRecord<Object, Object> producerRecord;
// allow for null keys
- if ( data.keyFieldIndex < 0 || r[ data.keyFieldIndex ] == null || StringUtils.isEmpty( r[ data.keyFieldIndex ].toString() ) ) {
+ if (data.keyFieldIndex < 0
+ || r[data.keyFieldIndex] == null
+ || StringUtils.isEmpty(r[data.keyFieldIndex].toString())) {
producerRecord = new ProducerRecord<>(resolve(meta.getTopic()), r[data.messageFieldIndex]);
} else {
- producerRecord = new ProducerRecord<>( resolve( meta.getTopic() ), r[ data.keyFieldIndex ],
+ producerRecord =
+ new ProducerRecord<>(
+ resolve(meta.getTopic()),
+ getInputRowMeta().getString(r, data.keyFieldIndex),
r[data.messageFieldIndex]);
}
