[skywalking-banyandb-java-client] 01/01: Drop invalid entries on writing.

[ha...@apache.org]

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

hanahmily pushed a commit to branch debug
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb-java-client.git

commit 3daf9cb0c8f1421220a9d7715fbbe23ccaa7d493
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Dec 29 21:39:05 2022 +0800

    Drop invalid entries on writing.
    
    Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
 .../banyandb/v1/client/AbstractBulkWriteProcessor.java | 17 ++++++++++++++---
 .../skywalking/banyandb/v1/client/AbstractWrite.java   | 18 ++++++++++++++++++
 2 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractBulkWriteProcessor.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractBulkWriteProcessor.java
index bc3aea8..5ccb885 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractBulkWriteProcessor.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractBulkWriteProcessor.java
@@ -23,6 +23,7 @@ import io.grpc.stub.AbstractAsyncStub;
 import io.grpc.stub.StreamObserver;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -66,11 +67,21 @@ public abstract class AbstractBulkWriteProcessor<REQ extends com.google.protobuf
         final StreamObserver<REQ> writeRequestStreamObserver
                 = this.buildStreamObserver(stub.withDeadlineAfter(flushInterval, TimeUnit.SECONDS), batch);
 
+        List sentData = new ArrayList(data.size());
         try {
             data.forEach(holder -> {
                 Holder h = (Holder) holder;
-                REQ request = ((AbstractWrite<REQ>) h.writeEntity()).build();
+                AbstractWrite<REQ> entity = (AbstractWrite<REQ>) h.writeEntity();
+                REQ request;
+                try {
+                    request = entity.build();
+                } catch (Throwable bt) {
+                    log.error("building the entity fails: {}", entity.toString(), bt);
+                    h.future().completeExceptionally(bt);
+                    return;
+                }
                 writeRequestStreamObserver.onNext(request);
+                sentData.add(h);
             });
         } catch (Throwable t) {
             log.error("Transform and send request to BanyanDB fail.", t);
@@ -80,12 +91,12 @@ public abstract class AbstractBulkWriteProcessor<REQ extends com.google.protobuf
         }
         batch.whenComplete((ignored, exp) -> {
             if (exp != null) {
-                data.stream().map((Function<Object, CompletableFuture<Void>>) o -> ((Holder) o).future())
+                sentData.stream().map((Function<Object, CompletableFuture<Void>>) o -> ((Holder) o).future())
                         .forEach((Consumer<CompletableFuture<Void>>) it -> it.completeExceptionally(exp));
                 log.error("Failed to execute requests in bulk", exp);
             } else {
                 log.debug("Succeeded to execute {} requests in bulk", data.size());
-                data.stream().map((Function<Object, CompletableFuture<Void>>) o -> ((Holder) o).future())
+                sentData.stream().map((Function<Object, CompletableFuture<Void>>) o -> ((Holder) o).future())
                         .forEach((Consumer<CompletableFuture<Void>>) it -> it.complete(null));
             }
         });
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractWrite.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractWrite.java
index 6ad76ca..4e07af9 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractWrite.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractWrite.java
@@ -19,6 +19,8 @@
 package org.apache.skywalking.banyandb.v1.client;
 
 import com.google.protobuf.Timestamp;
+
+import java.util.Map;
 import java.util.Optional;
 import lombok.Getter;
 import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
@@ -91,4 +93,20 @@ public abstract class AbstractWrite<P extends com.google.protobuf.GeneratedMessa
     }
 
     protected abstract P build(BanyandbCommon.Metadata metadata, Timestamp ts);
+
+    @Override
+    public String toString() {
+        StringBuilder stringBuilder = new StringBuilder();
+        stringBuilder.append("group=").append(group).append(", ").append("name=")
+                .append(name).append(", ").append("timestamp=").append(timestamp).append(", ");
+        for (int i = 0; i < this.tags.length; i++) {
+            final int index = i;
+            Map<String, MetadataCache.TagInfo> tagMap = this.entityMetadata.getTagOffset();
+            Optional<String> tagName = tagMap.keySet().stream().filter(name -> tagMap.get(name).getOffset() == index).findAny();
+            if (tagName.isPresent()) {
+                stringBuilder.append(tagName.get()).append("=").append(((Serializable<BanyandbModel.TagValue>) this.tags[i]).serialize()).append(", ");
+            }
+        }
+        return stringBuilder.toString();
+    }
 }