Add logging & ignore if payload is null before sending

Author: mzitnik

flink-connector-clickhouse-base/src/main/java/org/apache/flink/connector/clickhouse/convertor/ClickHouseConvertor.java

@@ -65,6 +65,7 @@ public ClickHousePayload apply( InputT o, SinkWriter.Context context) {
                 byte[] payload = this.pojoConvertor.convert(o);
                 return new ClickHousePayload(payload);
             } catch (Exception e) {
+                LOG.error("Failed to convert ClickHouse payload", e);
                 return new ClickHousePayload(null);
             }
         }

flink-connector-clickhouse-base/src/main/java/org/apache/flink/connector/clickhouse/sink/ClickHouseAsyncWriter.java

@@ -93,10 +93,12 @@ protected void submitRequestEntries(List<ClickHousePayload> requestEntries, Resu
         try {
             CompletableFuture<InsertResponse> response = chClient.insert(tableName, out -> {
                 for (ClickHousePayload requestEntry : requestEntries) {
-                    byte[] payload = requestEntry.getPayload();
-                    // sum the data that is sent to ClickHouse
-                    this.numBytesSendCounter.inc(payload.length);
-                    out.write(payload);
+                    if (requestEntry.getPayload() != null) {
+                        byte[] payload = requestEntry.getPayload();
+                        // sum the data that is sent to ClickHouse
+                        this.numBytesSendCounter.inc(payload.length);
+                        out.write(payload);
+                    }
                 }
                 // send the number that is sent to ClickHouse
                 this.numRecordsSendCounter.inc(requestEntries.size());