[pulsar] 05/12: [pulsar-io] Refine the key in redis sink when key is null (#11192)

[pe...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit eb5700d6987c7c6740be993550db15338d070274
Author: Fangbin Sun <su...@gmail.com>
AuthorDate: Wed Jul 14 14:51:15 2021 +0800

    [pulsar-io] Refine the key in redis sink when key is null (#11192)
    
    ### Motivation
    
    When the record has a null key, someone maybe confused by a NPE warn as below:
    
    ```
    11:26:22.730 [pool-5-thread-1] WARN  org.apache.pulsar.io.redis.sink.RedisSink - Record flush thread was exception
    java.lang.NullPointerException: null
            at java.util.concurrent.ConcurrentHashMap.putVal(ConcurrentHashMap.java:1011) ~[?:1.8.0_291]
            at java.util.concurrent.ConcurrentHashMap.put(ConcurrentHashMap.java:1006) ~[?:1.8.0_291]
            at org.apache.pulsar.io.redis.sink.RedisSink.flush(RedisSink.java:127) ~[pulsar-io-redis-2.8.0.nar-unpacked/:?]
            at org.apache.pulsar.io.redis.sink.RedisSink.lambda$write$1(RedisSink.java:94) ~[pulsar-io-redis-2.8.0.nar-unpacked/:?]
    ```
    
    We can use an empty string as key when its key is null as other sinks do.
    
    ### Modifications
    
    - Use an empty string as key when its key is null.
    
    (cherry picked from commit 27797d919975b89cff15d51ec50528fb31f37345)
---
 .../src/main/java/org/apache/pulsar/io/redis/sink/RedisSink.java     | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/pulsar-io/redis/src/main/java/org/apache/pulsar/io/redis/sink/RedisSink.java b/pulsar-io/redis/src/main/java/org/apache/pulsar/io/redis/sink/RedisSink.java
index 095f7a5..d1a7686 100644
--- a/pulsar-io/redis/src/main/java/org/apache/pulsar/io/redis/sink/RedisSink.java
+++ b/pulsar-io/redis/src/main/java/org/apache/pulsar/io/redis/sink/RedisSink.java
@@ -121,8 +121,9 @@ public class RedisSink implements Sink<byte[]> {
         if (CollectionUtils.isNotEmpty(recordsToFlush)) {
             for (Record<byte[]> record: recordsToFlush) {
                 try {
-                    // records with null keys or values will be ignored
-                    byte[] key = record.getKey().isPresent() ? record.getKey().get().getBytes(StandardCharsets.UTF_8) : null;
+                    // use an empty string as key when the key is null
+                    String recordKey = record.getKey().isPresent() ? record.getKey().get() : "";
+                    byte[] key = recordKey.getBytes(StandardCharsets.UTF_8);
                     byte[] value = record.getValue();
                     recordsToSet.put(key, value);
                 } catch (Exception e) {