fix: error converting from int to base64 for KafkaWorker (#79)

* fix: Kafka source worker file converting the variable into a string64 causes errors in certain case * fix: change key from a string to bye[]
vanus-labs · Oct 28, 2022 · 49b820f · 49b820f
1 parent 3ca4613
commit 49b820f
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 3 deletions.
diff --git a/connectors/source-kafka/src/main/java/com/linkall/source/kafka/KafkaData.java b/connectors/source-kafka/src/main/java/com/linkall/source/kafka/KafkaData.java
@@ -2,7 +2,7 @@
 
 import java.time.OffsetDateTime;
 
-public record KafkaData(String topic, String key, byte[] value, String KAFKA_SERVER_URL, OffsetDateTime timeStamp) {
+public record KafkaData(String topic, byte[] key, byte[] value, String KAFKA_SERVER_URL, OffsetDateTime timeStamp) {
 
 
 }
diff --git a/connectors/source-kafka/src/main/java/com/linkall/source/kafka/KafkaWorker.java b/connectors/source-kafka/src/main/java/com/linkall/source/kafka/KafkaWorker.java
@@ -76,8 +76,7 @@ public void doWork() {
 
             for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
                 OffsetDateTime timeStamp = new Date(record.timestamp()).toInstant().atOffset( ZoneOffset.UTC );
-                String key64 = Base64.getEncoder().encodeToString(record.key());
-                KafkaData kafkaData = new KafkaData(record.topic(), key64, record.value(), KAFKA_SERVER_URL, timeStamp);
+                KafkaData kafkaData = new KafkaData(record.topic(), record.key(), record.value(), KAFKA_SERVER_URL, timeStamp);
                 CloudEvent event = adapter.adapt(kafkaData);
                 String sink = ConfigUtil.getVanceSink();
                 System.out.println("message: " + Arrays.toString(record.value()));