You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2022/12/29 13:40:08 UTC
[skywalking-banyandb-java-client] branch debug created (now 3daf9cb)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch debug
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb-java-client.git
at 3daf9cb Drop invalid entries on writing.
This branch includes the following new commits:
new 3daf9cb Drop invalid entries on writing.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking-banyandb-java-client] 01/01: Drop invalid entries on writing.
Posted by 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();
+ }
}