commit a4570044b16091c6465eca46844594cce8eaa175
author Naveen Kumar <krnaveen14@gmail.com> Sat Nov 09 01:15:32 2019 +0530
committer Matteo Merli <mmerli@apache.org> Fri Nov 08 11:45:32 2019 -0800
tree 4665acbb9e1b4d1266bbc1a4b46b13b1819b70a0
parent 53ce33825208e3ec6437ab1e3f35032f5ca1b056

[pulsar-io] [kafka-connect-adapter] KafkaSourceRecord - Initialize key property with Optional.empty() if keyBytes from SourceRecord is null (Optional property should never itself be null) (#5591)

pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java

diff --git a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
index d9662c4..1bdd6c4 100644
--- a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
+++ b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
@@ -187,9 +187,7 @@
                 srcRecord.topic(), srcRecord.keySchema(), srcRecord.key());
             byte[] valueBytes = valueConverter.fromConnectData(
                 srcRecord.topic(), srcRecord.valueSchema(), srcRecord.value());
-            if (keyBytes != null) {
-                this.key = Optional.of(Base64.getEncoder().encodeToString(keyBytes));
-            }
+            this.key = keyBytes != null ? Optional.of(Base64.getEncoder().encodeToString(keyBytes)) : Optional.empty();
             this.value = new KeyValue(keyBytes, valueBytes);
 
             this.topicName = Optional.of(srcRecord.topic());