Kinesis-sink consider topic-name as partition-key if record key empty (…

…#2372)
apache · Aug 14, 2018 · ad5fc83 · ad5fc83
1 parent 7680173
commit ad5fc83
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
@@ -114,7 +114,7 @@ public void write(Record<byte[]> record) throws Exception {
                     record.getRecordSequence());
             throw new IllegalStateException("kinesis queue has publish failure");
         }
-        String partitionedKey = record.getKey().orElse(defaultPartitionedKey);
+        String partitionedKey = record.getKey().orElse(record.getTopicName().orElse(defaultPartitionedKey));
         partitionedKey = partitionedKey.length() > maxPartitionedKeyLength
                 ? partitionedKey.substring(0, maxPartitionedKeyLength - 1)
                 : partitionedKey; // partitionedKey Length must be at least one, and at most 256