Fix NPE while converting entity to Value (#51)

Author: lujiajing1126

src/main/java/org/apache/skywalking/banyandb/v1/client/Value.java

@@ -147,7 +147,10 @@ public static Serializable<BanyandbModel.TagValue> stringTagValue(String val) {
      * @param val payload
      * @return Anonymous tag with numeric payload
      */
-    public static Serializable<BanyandbModel.TagValue> longTagValue(long val) {
+    public static Serializable<BanyandbModel.TagValue> longTagValue(Long val) {
+        if (val == null) {
+            return nullTagValue();
+        }
         return new LongTagValue(val);
     }
 
@@ -158,6 +161,9 @@ public static Serializable<BanyandbModel.TagValue> longTagValue(long val) {
      * @return Anonymous tag with string array payload
      */
     public static Serializable<BanyandbModel.TagValue> stringArrayTagValue(List<String> val) {
+        if (val == null) {
+            return nullTagValue();
+        }
         return new StringArrayTagValue(val);
     }
 
@@ -168,6 +174,9 @@ public static Serializable<BanyandbModel.TagValue> stringArrayTagValue(List<Stri
      * @return Anonymous tag with binary payload
      */
     public static Serializable<BanyandbModel.TagValue> binaryTagValue(byte[] bytes) {
+        if (bytes == null) {
+            return nullTagValue();
+        }
         return new BinaryTagValue(ByteString.copyFrom(bytes));
     }
 
@@ -178,6 +187,9 @@ public static Serializable<BanyandbModel.TagValue> binaryTagValue(byte[] bytes)
      * @return Anonymous tag with numeric array payload
      */
     public static Serializable<BanyandbModel.TagValue> longArrayTag(List<Long> val) {
+        if (val == null) {
+            return nullTagValue();
+        }
         return new LongArrayTagValue(val);
     }
 

src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamWriteTest.java

@@ -166,11 +166,11 @@ public void onCompleted() {
         StreamWrite streamWrite = client.createStreamWrite("default", "sw", segmentId, now.toEpochMilli())
                 .tag("data_binary", Value.binaryTagValue(byteData))
                 .tag("trace_id", Value.stringTagValue(traceId)) // 0
-                .tag("state", Value.longTagValue(state)) // 1
+                .tag("state", Value.longTagValue((long) state)) // 1
                 .tag("service_id", Value.stringTagValue(serviceId)) // 2
                 .tag("service_instance_id", Value.stringTagValue(serviceInstanceId)) // 3
                 .tag("endpoint_id", Value.stringTagValue(endpointId)) // 4
-                .tag("duration", Value.longTagValue(latency)) // 5
+                .tag("duration", Value.longTagValue((long) latency)) // 5
                 .tag("http.method", Value.stringTagValue(null)) // 6
                 .tag("status_code", Value.stringTagValue(httpStatusCode)) // 7
                 .tag("db.type", Value.stringTagValue(dbType)) // 8
@@ -224,11 +224,11 @@ public void performSingleWrite() throws Exception {
         StreamWrite streamWrite = client.createStreamWrite("default", "sw", segmentId, now.toEpochMilli())
                 .tag("data_binary", Value.binaryTagValue(byteData))
                 .tag("trace_id", Value.stringTagValue(traceId)) // 0
-                .tag("state", Value.longTagValue(state)) // 1
+                .tag("state", Value.longTagValue((long) state)) // 1
                 .tag("service_id", Value.stringTagValue(serviceId)) // 2
                 .tag("service_instance_id", Value.stringTagValue(serviceInstanceId)) // 3
                 .tag("endpoint_id", Value.stringTagValue(endpointId)) // 4
-                .tag("duration", Value.longTagValue(latency)) // 5
+                .tag("duration", Value.longTagValue((long) latency)) // 5
                 .tag("http.method", Value.stringTagValue(null)) // 6
                 .tag("status_code", Value.stringTagValue(httpStatusCode)) // 7
                 .tag("db.type", Value.stringTagValue(dbType)) // 8
@@ -269,7 +269,7 @@ public void testAutoRefreshSchema() throws Exception {
         int latency = 200;
         int state = 1;
         Instant now = Instant.now();
-        byte[] byteData = new byte[] {14};
+        byte[] byteData = new byte[]{14};
         String broker = "172.16.10.129:9092";
         String topic = "topic_1";
         String queue = "queue_2";
@@ -280,11 +280,11 @@ public void testAutoRefreshSchema() throws Exception {
         StreamWrite streamWrite = client.createStreamWrite("default", "sw", segmentId, now.toEpochMilli())
                 .tag("data_binary", Value.binaryTagValue(byteData))
                 .tag("trace_id", Value.stringTagValue(traceId)) // 0
-                .tag("state", Value.longTagValue(state)) // 1
+                .tag("state", Value.longTagValue((long) state)) // 1
                 .tag("service_id", Value.stringTagValue(serviceId)) // 2
                 .tag("service_instance_id", Value.stringTagValue(serviceInstanceId)) // 3
                 .tag("endpoint_id", Value.stringTagValue(endpointId)) // 4
-                .tag("duration", Value.longTagValue(latency)) // 5
+                .tag("duration", Value.longTagValue((long) latency)) // 5
                 .tag("http.method", Value.stringTagValue(null)) // 6
                 .tag("status_code", Value.stringTagValue(httpStatusCode)) // 7
                 .tag("db.type", Value.stringTagValue(dbType)) // 8
@@ -320,11 +320,11 @@ public void testAutoRefreshSchema() throws Exception {
         streamWrite = client.createStreamWrite("default", "sw", segmentId, now.toEpochMilli())
                 .tag("data_binary", Value.binaryTagValue(byteData))
                 .tag("trace_id", Value.stringTagValue(traceId)) // 0
-                .tag("state", Value.longTagValue(state)) // 1
+                .tag("state", Value.longTagValue((long) state)) // 1
                 .tag("service_id", Value.stringTagValue(serviceId)) // 2
                 .tag("service_instance_id", Value.stringTagValue(serviceInstanceId)) // 3
                 .tag("endpoint_id", Value.stringTagValue(endpointId)) // 4
-                .tag("duration", Value.longTagValue(latency)) // 5
+                .tag("duration", Value.longTagValue((long) latency)) // 5
                 .tag("http.method", Value.stringTagValue(null)) // 6
                 .tag("status_code", Value.stringTagValue(httpStatusCode)) // 7
                 .tag("db.type", Value.stringTagValue(dbType)) // 8