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 2023/09/17 23:14:18 UTC
[skywalking-banyandb-java-client] branch main updated: Add mod revision check to write requests(measure/stream) (#46)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb-java-client.git
The following commit(s) were added to refs/heads/main by this push:
new 32eae17 Add mod revision check to write requests(measure/stream) (#46)
32eae17 is described below
commit 32eae1725fc54d9c9b0307643f07d04ad56c567a
Author: hailin0 <ha...@gmail.com>
AuthorDate: Mon Sep 18 07:14:14 2023 +0800
Add mod revision check to write requests(measure/stream) (#46)
* Add mod revision check to write requests(measure/stream)
- Support cached mod revision when create/update schema
- Support auto-refresh mod revision when writing data
---
CHANGES.md | 7 +
README.md | 6 +-
.../banyandb/v1/client/AbstractWrite.java | 2 +-
.../banyandb/v1/client/BanyanDBClient.java | 89 ++++++++++-
.../v1/client/MeasureBulkWriteProcessor.java | 42 ++++-
.../banyandb/v1/client/MeasureWrite.java | 1 +
.../v1/client/StreamBulkWriteProcessor.java | 33 +++-
.../skywalking/banyandb/v1/client/StreamWrite.java | 1 +
.../banyandb/v1/client/grpc/MetadataClient.java | 5 +-
.../v1/client/metadata/GroupMetadataRegistry.java | 3 +-
.../metadata/IndexRuleBindingMetadataRegistry.java | 3 +-
.../client/metadata/IndexRuleMetadataRegistry.java | 3 +-
.../banyandb/v1/client/metadata/Measure.java | 26 +++-
.../client/metadata/MeasureMetadataRegistry.java | 5 +-
.../banyandb/v1/client/metadata/MetadataCache.java | 13 +-
.../banyandb/v1/client/metadata/NamedSchema.java | 9 ++
.../banyandb/v1/client/metadata/Stream.java | 28 +++-
.../v1/client/metadata/StreamMetadataRegistry.java | 5 +-
.../metadata/TopNAggregationMetadataRegistry.java | 3 +-
src/main/proto/banyandb/v1/banyandb-database.proto | 4 +
src/main/proto/banyandb/v1/banyandb-measure.proto | 11 +-
src/main/proto/banyandb/v1/banyandb-model.proto | 10 ++
src/main/proto/banyandb/v1/banyandb-stream.proto | 13 +-
.../v1/client/AbstractBanyanDBClientTest.java | 25 ++-
.../v1/client/BanyanDBClientMeasureWriteTest.java | 132 +++++++++++++---
.../v1/client/BanyanDBClientStreamWriteTest.java | 173 ++++++++++++++++++---
.../banyandb/v1/client/BanyanDBClientTestCI.java | 2 +-
.../metadata/MeasureMetadataRegistryTest.java | 4 +-
.../metadata/StreamMetadataRegistryTest.java | 4 +-
29 files changed, 568 insertions(+), 94 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 2ef214e..8f80e2a 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -2,6 +2,13 @@ Changes by Version
==================
Release Notes.
+0.5.0
+------------------
+
+### Features
+
+* Add mod revision check to write requests
+
0.4.0
------------------
diff --git a/README.md b/README.md
index 895ef24..4da8175 100644
--- a/README.md
+++ b/README.md
@@ -81,7 +81,7 @@ Measure m = Measure.create("sw_metric", "service_cpm_minute", Duration.ofHours(1
.setEntityRelativeTags("entity_id")
// define a tag family "default"
.addTagFamily(TagFamilySpec.create("default")
- .addTagSpec(TagFamilySpec.TagSpec.newIDTag("id"))
+ .addTagSpec(TagFamilySpec.TagSpec.newStringTag("id"))
.addTagSpec(TagFamilySpec.TagSpec.newStringTag("entity_id"))
.build())
// define field specs
@@ -280,7 +280,7 @@ The writing procedure for `Measure` is similar to the above described process an
```java
// build a MeasureBulkWriteProcessor from client
-MeasureBulkWriteProcessor bulkWriteProcessor = client.buildMeasureWriteProcessor(maxBulkSize, flushInterval, concurrency);
+MeasureBulkWriteProcessor measureBulkWriteProcessor = client.buildMeasureWriteProcessor(maxBulkSize, flushInterval, concurrency);
```
A `BulkWriteProcessor` is created by calling `buildMeasureWriteProcessor`. Then build the `MeasureWrite` object and send with bulk processor,
@@ -288,7 +288,7 @@ A `BulkWriteProcessor` is created by calling `buildMeasureWriteProcessor`. Then
```java
Instant now = Instant.now();
MeasureWrite measureWrite = client.createMeasureWrite("sw_metric", "service_cpm_minute", now.toEpochMilli());
- measureWrite.tag("id", TagAndValue.idTagValue("1"))
+ measureWrite.tag("id", TagAndValue.stringTagValue("1"))
.tag("entity_id", TagAndValue.stringTagValue("entity_1"))
.field("total", TagAndValue.longFieldValue(100))
.field("value", TagAndValue.longFieldValue(1));
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 dd317d7..72067da 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
@@ -74,7 +74,7 @@ public abstract class AbstractWrite<P extends com.google.protobuf.GeneratedMessa
}
BanyandbCommon.Metadata metadata = BanyandbCommon.Metadata.newBuilder()
- .setGroup(entityMetadata.getGroup()).setName(entityMetadata.getName()).build();
+ .setGroup(entityMetadata.getGroup()).setName(entityMetadata.getName()).setModRevision(entityMetadata.getModRevision()).build();
Timestamp ts = Timestamp.newBuilder()
.setSeconds(timestamp / 1000)
.setNanos((int) (timestamp % 1000 * 1_000_000)).build();
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClient.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClient.java
index 8f68f31..bd33d8f 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClient.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClient.java
@@ -28,6 +28,8 @@ import io.grpc.stub.StreamObserver;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.measure.v1.BanyandbMeasure;
import org.apache.skywalking.banyandb.measure.v1.MeasureServiceGrpc;
import org.apache.skywalking.banyandb.stream.v1.BanyandbStream;
@@ -36,6 +38,8 @@ import org.apache.skywalking.banyandb.v1.client.grpc.HandleExceptionsWith;
import org.apache.skywalking.banyandb.v1.client.grpc.channel.ChannelManager;
import org.apache.skywalking.banyandb.v1.client.grpc.channel.DefaultChannelFactory;
import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
+import org.apache.skywalking.banyandb.v1.client.grpc.exception.InternalException;
+import org.apache.skywalking.banyandb.v1.client.grpc.exception.InvalidArgumentException;
import org.apache.skywalking.banyandb.v1.client.metadata.Group;
import org.apache.skywalking.banyandb.v1.client.metadata.GroupMetadataRegistry;
import org.apache.skywalking.banyandb.v1.client.metadata.IndexRule;
@@ -58,6 +62,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
@@ -198,26 +203,65 @@ public class BanyanDBClient implements Closeable {
* Perform a single write with given entity.
*
* @param streamWrite the entity to be written
+ * @return a future of write result
*/
- public void write(StreamWrite streamWrite) {
+ public CompletableFuture<Void> write(StreamWrite streamWrite) {
checkState(this.streamServiceStub != null, "stream service is null");
+ CompletableFuture<Void> future = new CompletableFuture<>();
final StreamObserver<BanyandbStream.WriteRequest> writeRequestStreamObserver
= this.streamServiceStub
.withDeadlineAfter(this.getOptions().getDeadline(), TimeUnit.SECONDS)
.write(
new StreamObserver<BanyandbStream.WriteResponse>() {
+ private BanyanDBException responseException;
+
@Override
public void onNext(BanyandbStream.WriteResponse writeResponse) {
+ switch (writeResponse.getStatus()) {
+ case STATUS_INVALID_TIMESTAMP:
+ responseException = new InvalidArgumentException(
+ "Invalid timestamp: " + streamWrite.getTimestamp(), null, Status.Code.INVALID_ARGUMENT, false);
+ break;
+ case STATUS_NOT_FOUND:
+ responseException = new InvalidArgumentException(
+ "Invalid metadata: " + streamWrite.entityMetadata, null, Status.Code.INVALID_ARGUMENT, false);
+ break;
+ case STATUS_EXPIRED_SCHEMA:
+ BanyandbCommon.Metadata metadata = writeResponse.getMetadata();
+ log.warn("The schema {}.{} is expired, trying update the schema...",
+ metadata.getGroup(), metadata.getName());
+ try {
+ BanyanDBClient.this.findStream(metadata.getGroup(), metadata.getName());
+ } catch (BanyanDBException e) {
+ String warnMessage = String.format("Failed to refresh the stream schema %s.%s",
+ metadata.getGroup(), metadata.getName());
+ log.warn(warnMessage, e);
+ }
+ responseException = new InvalidArgumentException(
+ "Expired revision: " + metadata.getModRevision(), null, Status.Code.INVALID_ARGUMENT, true);
+ break;
+ case STATUS_INTERNAL_ERROR:
+ responseException = new InternalException(
+ "Internal error occurs in server", null, Status.Code.INTERNAL, true);
+ break;
+ default:
+ }
}
@Override
public void onError(Throwable throwable) {
log.error("Error occurs in flushing streams.", throwable);
+ future.completeExceptionally(throwable);
}
@Override
public void onCompleted() {
+ if (responseException == null) {
+ future.complete(null);
+ } else {
+ future.completeExceptionally(responseException);
+ }
}
});
try {
@@ -225,6 +269,7 @@ public class BanyanDBClient implements Closeable {
} finally {
writeRequestStreamObserver.onCompleted();
}
+ return future;
}
/**
@@ -239,7 +284,7 @@ public class BanyanDBClient implements Closeable {
public StreamBulkWriteProcessor buildStreamWriteProcessor(int maxBulkSize, int flushInterval, int concurrency) {
checkState(this.streamServiceStub != null, "stream service is null");
- return new StreamBulkWriteProcessor(this.streamServiceStub, maxBulkSize, flushInterval, concurrency);
+ return new StreamBulkWriteProcessor(this, maxBulkSize, flushInterval, concurrency);
}
/**
@@ -254,7 +299,7 @@ public class BanyanDBClient implements Closeable {
public MeasureBulkWriteProcessor buildMeasureWriteProcessor(int maxBulkSize, int flushInterval, int concurrency) {
checkState(this.measureServiceStub != null, "measure service is null");
- return new MeasureBulkWriteProcessor(this.measureServiceStub, maxBulkSize, flushInterval, concurrency);
+ return new MeasureBulkWriteProcessor(this, maxBulkSize, flushInterval, concurrency);
}
/**
@@ -361,11 +406,28 @@ public class BanyanDBClient implements Closeable {
*/
public void define(Stream stream) throws BanyanDBException {
StreamMetadataRegistry streamRegistry = new StreamMetadataRegistry(checkNotNull(this.channel));
- streamRegistry.create(stream);
+ long modRevision = streamRegistry.create(stream);
defineIndexRules(stream, stream.indexRules());
+
+ stream = stream.withModRevision(modRevision);
this.metadataCache.register(stream);
}
+ /**
+ * Delete a stream
+ *
+ * @param stream the stream to be deleted
+ * @return true if the stream is deleted successfully
+ */
+ public boolean delete(Stream stream) throws BanyanDBException {
+ StreamMetadataRegistry streamRegistry = new StreamMetadataRegistry(checkNotNull(this.channel));
+ if (streamRegistry.delete(stream.group(), stream.name())) {
+ this.metadataCache.unregister(stream);
+ return true;
+ }
+ return false;
+ }
+
/**
* Define a new measure
*
@@ -373,11 +435,28 @@ public class BanyanDBClient implements Closeable {
*/
public void define(Measure measure) throws BanyanDBException {
MeasureMetadataRegistry measureRegistry = new MeasureMetadataRegistry(checkNotNull(this.channel));
- measureRegistry.create(measure);
+ long modRevision = measureRegistry.create(measure);
defineIndexRules(measure, measure.indexRules());
+
+ measure = measure.withModRevision(modRevision);
this.metadataCache.register(measure);
}
+ /**
+ * Delete a measure
+ *
+ * @param measure the measure to be deleted
+ * @return true if the measure is deleted successfully
+ */
+ public boolean delete(Measure measure) throws BanyanDBException {
+ MeasureMetadataRegistry measureRegistry = new MeasureMetadataRegistry(checkNotNull(this.channel));
+ if (measureRegistry.delete(measure.group(), measure.name())) {
+ this.metadataCache.unregister(measure);
+ return true;
+ }
+ return false;
+ }
+
/**
* Define a new TopNAggregation
*
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureBulkWriteProcessor.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureBulkWriteProcessor.java
index 690e2e3..5d3ce23 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureBulkWriteProcessor.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureBulkWriteProcessor.java
@@ -20,10 +20,16 @@ package org.apache.skywalking.banyandb.v1.client;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
+
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.measure.v1.BanyandbMeasure;
import org.apache.skywalking.banyandb.measure.v1.MeasureServiceGrpc;
+import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.HashSet;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
/**
@@ -33,30 +39,50 @@ import java.util.concurrent.CompletableFuture;
@ThreadSafe
public class MeasureBulkWriteProcessor extends AbstractBulkWriteProcessor<BanyandbMeasure.WriteRequest,
MeasureServiceGrpc.MeasureServiceStub> {
+ private final BanyanDBClient client;
+
/**
* Create the processor.
*
- * @param measureServiceStub stub for gRPC call.
- * @param maxBulkSize the max bulk size for the flush operation
- * @param flushInterval if given maxBulkSize is not reached in this period, the flush would be trigger
- * automatically. Unit is second.
- * @param concurrency the number of concurrency would run for the flush max.
+ * @param client the client
+ * @param maxBulkSize the max bulk size for the flush operation
+ * @param flushInterval if given maxBulkSize is not reached in this period, the flush would be trigger
+ * automatically. Unit is second.
+ * @param concurrency the number of concurrency would run for the flush max.
*/
protected MeasureBulkWriteProcessor(
- final MeasureServiceGrpc.MeasureServiceStub measureServiceStub,
+ final BanyanDBClient client,
final int maxBulkSize,
final int flushInterval,
final int concurrency) {
- super(measureServiceStub, "MeasureBulkWriteProcessor", maxBulkSize, flushInterval, concurrency);
+ super(client.getMeasureServiceStub(), "MeasureBulkWriteProcessor", maxBulkSize, flushInterval, concurrency);
+ this.client = client;
}
@Override
protected StreamObserver<BanyandbMeasure.WriteRequest> buildStreamObserver(MeasureServiceGrpc.MeasureServiceStub stub,
CompletableFuture<Void> batch) {
return stub.write(new StreamObserver<BanyandbMeasure.WriteResponse>() {
+ private final Set<String> schemaExpired = new HashSet<>();
+
@Override
public void onNext(BanyandbMeasure.WriteResponse writeResponse) {
-
+ switch (writeResponse.getStatus()) {
+ case STATUS_EXPIRED_SCHEMA:
+ BanyandbCommon.Metadata metadata = writeResponse.getMetadata();
+ String schemaKey = metadata.getGroup() + "." + metadata.getName();
+ if (!schemaExpired.contains(schemaKey)) {
+ log.warn("The schema {} is expired, trying update the schema...", schemaKey);
+ try {
+ client.findMeasure(metadata.getGroup(), metadata.getName());
+ schemaExpired.add(schemaKey);
+ } catch (BanyanDBException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ break;
+ default:
+ }
}
@Override
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureWrite.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureWrite.java
index 7510128..d405c61 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureWrite.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureWrite.java
@@ -93,6 +93,7 @@ public class MeasureWrite extends AbstractWrite<BanyandbMeasure.WriteRequest> {
}
builder.setDataPoint(datapointValueBuilder);
+ builder.setMessageId(System.nanoTime());
return builder.build();
}
}
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamBulkWriteProcessor.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamBulkWriteProcessor.java
index bcb4f9e..2dd6a33 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamBulkWriteProcessor.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamBulkWriteProcessor.java
@@ -21,10 +21,16 @@ package org.apache.skywalking.banyandb.v1.client;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
+
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.stream.v1.StreamServiceGrpc;
import org.apache.skywalking.banyandb.stream.v1.BanyandbStream;
+import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.HashSet;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
/**
@@ -34,29 +40,50 @@ import java.util.concurrent.CompletableFuture;
@ThreadSafe
public class StreamBulkWriteProcessor extends AbstractBulkWriteProcessor<BanyandbStream.WriteRequest,
StreamServiceGrpc.StreamServiceStub> {
+ private final BanyanDBClient client;
+
/**
* Create the processor.
*
- * @param serviceStub stub for gRPC call.
+ * @param client the client
* @param maxBulkSize the max bulk size for the flush operation
* @param flushInterval if given maxBulkSize is not reached in this period, the flush would be trigger
* automatically. Unit is second.
* @param concurrency the number of concurrency would run for the flush max.
*/
protected StreamBulkWriteProcessor(
- final StreamServiceGrpc.StreamServiceStub serviceStub,
+ final BanyanDBClient client,
final int maxBulkSize,
final int flushInterval,
final int concurrency) {
- super(serviceStub, "StreamBulkWriteProcessor", maxBulkSize, flushInterval, concurrency);
+ super(client.getStreamServiceStub(), "StreamBulkWriteProcessor", maxBulkSize, flushInterval, concurrency);
+ this.client = client;
}
@Override
protected StreamObserver<BanyandbStream.WriteRequest> buildStreamObserver(StreamServiceGrpc.StreamServiceStub stub, CompletableFuture<Void> batch) {
return stub.write(
new StreamObserver<BanyandbStream.WriteResponse>() {
+ private final Set<String> schemaExpired = new HashSet<>();
+
@Override
public void onNext(BanyandbStream.WriteResponse writeResponse) {
+ switch (writeResponse.getStatus()) {
+ case STATUS_EXPIRED_SCHEMA:
+ BanyandbCommon.Metadata metadata = writeResponse.getMetadata();
+ String schemaKey = metadata.getGroup() + "." + metadata.getName();
+ if (!schemaExpired.contains(schemaKey)) {
+ log.warn("The schema {} is expired, trying update the schema...", schemaKey);
+ try {
+ client.findStream(metadata.getGroup(), metadata.getName());
+ schemaExpired.add(schemaKey);
+ } catch (BanyanDBException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ break;
+ default:
+ }
}
@Override
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamWrite.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamWrite.java
index a9f3c2c..d77b26e 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamWrite.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamWrite.java
@@ -94,6 +94,7 @@ public class StreamWrite extends AbstractWrite<BanyandbStream.WriteRequest> {
elemValBuilder.addTagFamilies(BanyandbModel.TagFamilyForWrite.newBuilder().addAllTags(tags).build());
}
builder.setElement(elemValBuilder);
+ builder.setMessageId(System.nanoTime());
return builder.build();
}
}
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/grpc/MetadataClient.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/grpc/MetadataClient.java
index d64dc11..e820e55 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/grpc/MetadataClient.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/grpc/MetadataClient.java
@@ -33,6 +33,8 @@ import java.util.List;
* @param <S> NamedSchema: Java implementation (POJO) which can be serialized to P
*/
public abstract class MetadataClient<STUB extends AbstractBlockingStub<STUB>, P extends GeneratedMessageV3, S extends NamedSchema<P>> {
+ public static final long DEFAULT_MOD_REVISION = 0;
+
protected final STUB stub;
protected MetadataClient(STUB stub) {
@@ -43,9 +45,10 @@ public abstract class MetadataClient<STUB extends AbstractBlockingStub<STUB>, P
* Create a schema
*
* @param payload the schema to be created
+ * @return the mod revision of the schema
* @throws BanyanDBException a wrapped exception to the underlying gRPC calls
*/
- public abstract void create(S payload) throws BanyanDBException;
+ public abstract long create(S payload) throws BanyanDBException;
/**
* Update the schema
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/GroupMetadataRegistry.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/GroupMetadataRegistry.java
index 6d73dcb..ac8204b 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/GroupMetadataRegistry.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/GroupMetadataRegistry.java
@@ -36,10 +36,11 @@ public class GroupMetadataRegistry extends MetadataClient<GroupRegistryServiceGr
}
@Override
- public void create(Group payload) throws BanyanDBException {
+ public long create(Group payload) throws BanyanDBException {
execute(() -> stub.create(BanyandbDatabase.GroupRegistryServiceCreateRequest.newBuilder()
.setGroup(payload.serialize())
.build()));
+ return DEFAULT_MOD_REVISION;
}
@Override
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/IndexRuleBindingMetadataRegistry.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/IndexRuleBindingMetadataRegistry.java
index 5674a83..9f219e9 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/IndexRuleBindingMetadataRegistry.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/IndexRuleBindingMetadataRegistry.java
@@ -36,10 +36,11 @@ public class IndexRuleBindingMetadataRegistry extends MetadataClient<IndexRuleBi
}
@Override
- public void create(IndexRuleBinding payload) throws BanyanDBException {
+ public long create(IndexRuleBinding payload) throws BanyanDBException {
execute(() -> stub.create(BanyandbDatabase.IndexRuleBindingRegistryServiceCreateRequest.newBuilder()
.setIndexRuleBinding(payload.serialize())
.build()));
+ return DEFAULT_MOD_REVISION;
}
@Override
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/IndexRuleMetadataRegistry.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/IndexRuleMetadataRegistry.java
index 81f8543..d1c2e47 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/IndexRuleMetadataRegistry.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/IndexRuleMetadataRegistry.java
@@ -35,11 +35,12 @@ public class IndexRuleMetadataRegistry extends MetadataClient<IndexRuleRegistryS
}
@Override
- public void create(IndexRule payload) throws BanyanDBException {
+ public long create(IndexRule payload) throws BanyanDBException {
execute(() ->
stub.create(BanyandbDatabase.IndexRuleRegistryServiceCreateRequest.newBuilder()
.setIndexRule(payload.serialize())
.build()));
+ return DEFAULT_MOD_REVISION;
}
@Override
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Measure.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Measure.java
index 1334dcd..d14fe1c 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Measure.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Measure.java
@@ -20,10 +20,15 @@ package org.apache.skywalking.banyandb.v1.client.metadata;
import com.google.auto.value.AutoValue;
import com.google.common.collect.ImmutableList;
+
+import javax.annotation.Nullable;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
+
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
+import org.apache.skywalking.banyandb.v1.client.util.IgnoreHashEquals;
import org.apache.skywalking.banyandb.v1.client.util.TimeUtils;
import java.time.ZonedDateTime;
@@ -53,6 +58,14 @@ public abstract class Measure extends NamedSchema<BanyandbDatabase.Measure> {
*/
abstract Duration interval();
+ /**
+ * last updated revision
+ * This field can only be set by the server.
+ */
+ @Nullable
+ @IgnoreHashEquals
+ abstract Long modRevision();
+
/**
* index rules bound to the stream
*/
@@ -65,6 +78,10 @@ public abstract class Measure extends NamedSchema<BanyandbDatabase.Measure> {
.build();
}
+ public final Measure withModRevision(Long modRevision) {
+ return toBuilder().setModRevision(modRevision).build();
+ }
+
public static Measure.Builder create(String group, String name, Duration interval) {
return new AutoValue_Measure.Builder().setGroup(group).setName(name).setInterval(interval);
}
@@ -81,6 +98,8 @@ public abstract class Measure extends NamedSchema<BanyandbDatabase.Measure> {
abstract Measure.Builder setUpdatedAt(ZonedDateTime updatedAt);
+ abstract Measure.Builder setModRevision(Long modRevision);
+
abstract ImmutableList.Builder<TagFamilySpec> tagFamiliesBuilder();
public final Measure.Builder addTagFamily(TagFamilySpec tagFamilySpec) {
@@ -127,6 +146,7 @@ public abstract class Measure extends NamedSchema<BanyandbDatabase.Measure> {
final Measure.Builder m = Measure.create(pb.getMetadata().getGroup(), pb.getMetadata().getName(),
Duration.parse(pb.getInterval()))
.setUpdatedAt(TimeUtils.parseTimestamp(pb.getUpdatedAt()))
+ .setModRevision(pb.getMetadata().getModRevision())
.setEntityRelativeTags(pb.getEntity().getTagNamesList());
// build tag family spec
@@ -154,9 +174,13 @@ public abstract class Measure extends NamedSchema<BanyandbDatabase.Measure> {
fs.add(spec.serialize());
}
+ BanyandbCommon.Metadata metadata = buildMetadata();
+ if (this.modRevision() != null) {
+ metadata = buildMetadata(this.modRevision());
+ }
BanyandbDatabase.Measure.Builder b = BanyandbDatabase.Measure.newBuilder()
.setInterval(interval().format())
- .setMetadata(buildMetadata())
+ .setMetadata(metadata)
.addAllTagFamilies(tfs)
.addAllFields(fs)
.setEntity(BanyandbDatabase.Entity.newBuilder().addAllTagNames(entityRelativeTags()).build());
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/MeasureMetadataRegistry.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/MeasureMetadataRegistry.java
index 7937c76..f378156 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/MeasureMetadataRegistry.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/MeasureMetadataRegistry.java
@@ -36,11 +36,12 @@ public class MeasureMetadataRegistry extends MetadataClient<MeasureRegistryServi
}
@Override
- public void create(final Measure payload) throws BanyanDBException {
- execute(() ->
+ public long create(final Measure payload) throws BanyanDBException {
+ BanyandbDatabase.MeasureRegistryServiceCreateResponse resp = execute(() ->
stub.create(BanyandbDatabase.MeasureRegistryServiceCreateRequest.newBuilder()
.setMeasure(payload.serialize())
.build()));
+ return resp.getModRevision();
}
@Override
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/MetadataCache.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/MetadataCache.java
index e57c280..fa838e3 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/MetadataCache.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/MetadataCache.java
@@ -39,11 +39,19 @@ public class MetadataCache {
return stream;
}
+ public EntityMetadata unregister(Stream stream) {
+ return this.cache.remove(formatKey(stream.group(), stream.name()));
+ }
+
public Measure register(Measure measure) {
this.cache.put(formatKey(measure.group(), measure.name()), parse(measure));
return measure;
}
+ public EntityMetadata unregister(Measure measure) {
+ return this.cache.remove(formatKey(measure.group(), measure.name()));
+ }
+
public EntityMetadata findMetadata(String group, String name) {
return this.cache.get(formatKey(group, name));
}
@@ -65,7 +73,7 @@ public class MetadataCache {
tagInfo.put(s.tagFamilies().get(i).tagSpecs().get(j).getTagName(), new TagInfo(tagFamilyName, k++));
}
}
- return new EntityMetadata(s.group(), s.name(), totalTags, 0, tagFamilyCapacity,
+ return new EntityMetadata(s.group(), s.name(), s.modRevision(), totalTags, 0, tagFamilyCapacity,
Collections.unmodifiableMap(tagInfo),
Collections.emptyMap());
}
@@ -87,7 +95,7 @@ public class MetadataCache {
for (int i = 0; i < m.fields().size(); i++) {
fieldOffset.put(m.fields().get(i).getName(), i);
}
- return new EntityMetadata(m.group(), m.name(), totalTags, m.fields().size(), tagFamilyCapacity,
+ return new EntityMetadata(m.group(), m.name(), m.modRevision(), totalTags, m.fields().size(), tagFamilyCapacity,
Collections.unmodifiableMap(tagOffset), Collections.unmodifiableMap(fieldOffset));
}
@@ -96,6 +104,7 @@ public class MetadataCache {
public static class EntityMetadata {
private final String group;
private final String name;
+ private final long modRevision;
private final int totalTags;
private final int totalFields;
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/NamedSchema.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/NamedSchema.java
index 660eea0..5e3ab2d 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/NamedSchema.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/NamedSchema.java
@@ -60,4 +60,13 @@ public abstract class NamedSchema<P extends GeneratedMessageV3> {
return BanyandbCommon.Metadata.newBuilder().setName(name()).setGroup(group()).build();
}
}
+
+ protected BanyandbCommon.Metadata buildMetadata(long modRevision) {
+ BanyandbCommon.Metadata.Builder builder =
+ BanyandbCommon.Metadata.newBuilder().setName(name()).setModRevision(modRevision);
+ if (!Strings.isNullOrEmpty(group())) {
+ builder.setGroup(group());
+ }
+ return builder.build();
+ }
}
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Stream.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Stream.java
index 88864fb..252dc1d 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Stream.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/Stream.java
@@ -20,9 +20,14 @@ package org.apache.skywalking.banyandb.v1.client.metadata;
import com.google.auto.value.AutoValue;
import com.google.common.collect.ImmutableList;
+
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
+import org.apache.skywalking.banyandb.v1.client.util.IgnoreHashEquals;
import org.apache.skywalking.banyandb.v1.client.util.TimeUtils;
+import javax.annotation.Nullable;
+
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.List;
@@ -39,17 +44,29 @@ public abstract class Stream extends NamedSchema<BanyandbDatabase.Stream> {
*/
abstract ImmutableList<String> entityRelativeTags();
+ /**
+ * last updated revision
+ * This field can only be set by the server.
+ */
+ @Nullable
+ @IgnoreHashEquals
+ abstract Long modRevision();
+
/**
* index rules bound to the stream
*/
public abstract ImmutableList<IndexRule> indexRules();
- abstract Builder toBuilder();
+ public abstract Builder toBuilder();
public final Stream withIndexRules(List<IndexRule> indexRules) {
return toBuilder().addIndexes(indexRules).build();
}
+ public final Stream withModRevision(long modRevision) {
+ return toBuilder().setModRevision(modRevision).build();
+ }
+
public static Stream.Builder create(String group, String name) {
return new AutoValue_Stream.Builder().setGroup(group).setName(name);
}
@@ -64,6 +81,8 @@ public abstract class Stream extends NamedSchema<BanyandbDatabase.Stream> {
abstract Builder setUpdatedAt(ZonedDateTime updatedAt);
+ abstract Builder setModRevision(Long modRevision);
+
abstract ImmutableList.Builder<TagFamilySpec> tagFamiliesBuilder();
public final Builder addTagFamily(TagFamilySpec tagFamilySpec) {
@@ -104,8 +123,12 @@ public abstract class Stream extends NamedSchema<BanyandbDatabase.Stream> {
metadataTagFamilySpecs.add(spec.serialize());
}
+ BanyandbCommon.Metadata metadata = buildMetadata();
+ if (this.modRevision() != null) {
+ metadata = buildMetadata(this.modRevision());
+ }
BanyandbDatabase.Stream.Builder b = BanyandbDatabase.Stream.newBuilder()
- .setMetadata(buildMetadata())
+ .setMetadata(metadata)
.addAllTagFamilies(metadataTagFamilySpecs)
.setEntity(BanyandbDatabase.Entity.newBuilder().addAllTagNames(entityRelativeTags()).build());
@@ -118,6 +141,7 @@ public abstract class Stream extends NamedSchema<BanyandbDatabase.Stream> {
public static Stream fromProtobuf(final BanyandbDatabase.Stream pb) {
Stream.Builder s = Stream.create(pb.getMetadata().getGroup(), pb.getMetadata().getName())
.setUpdatedAt(TimeUtils.parseTimestamp(pb.getUpdatedAt()))
+ .setModRevision(pb.getMetadata().getModRevision())
.setEntityRelativeTags(pb.getEntity().getTagNamesList());
// build tag family spec
for (int i = 0; i < pb.getTagFamiliesCount(); i++) {
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/StreamMetadataRegistry.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/StreamMetadataRegistry.java
index d59065f..cde4b3e 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/StreamMetadataRegistry.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/StreamMetadataRegistry.java
@@ -36,11 +36,12 @@ public class StreamMetadataRegistry extends MetadataClient<StreamRegistryService
}
@Override
- public void create(Stream payload) throws BanyanDBException {
- execute(() ->
+ public long create(Stream payload) throws BanyanDBException {
+ BanyandbDatabase.StreamRegistryServiceCreateResponse resp = execute(() ->
stub.create(BanyandbDatabase.StreamRegistryServiceCreateRequest.newBuilder()
.setStream(payload.serialize())
.build()));
+ return resp.getModRevision();
}
@Override
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/TopNAggregationMetadataRegistry.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/TopNAggregationMetadataRegistry.java
index d10e5d7..4204f04 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/TopNAggregationMetadataRegistry.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/TopNAggregationMetadataRegistry.java
@@ -35,11 +35,12 @@ public class TopNAggregationMetadataRegistry extends MetadataClient<TopNAggregat
}
@Override
- public void create(TopNAggregation payload) throws BanyanDBException {
+ public long create(TopNAggregation payload) throws BanyanDBException {
execute(() ->
stub.create(BanyandbDatabase.TopNAggregationRegistryServiceCreateRequest.newBuilder()
.setTopNAggregation(payload.serialize())
.build()));
+ return DEFAULT_MOD_REVISION;
}
@Override
diff --git a/src/main/proto/banyandb/v1/banyandb-database.proto b/src/main/proto/banyandb/v1/banyandb-database.proto
index 6e1004c..c532df2 100644
--- a/src/main/proto/banyandb/v1/banyandb-database.proto
+++ b/src/main/proto/banyandb/v1/banyandb-database.proto
@@ -211,6 +211,7 @@ message StreamRegistryServiceCreateRequest {
}
message StreamRegistryServiceCreateResponse {
+ int64 mod_revision = 1;
}
message StreamRegistryServiceUpdateRequest {
@@ -218,6 +219,7 @@ message StreamRegistryServiceUpdateRequest {
}
message StreamRegistryServiceUpdateResponse {
+ int64 mod_revision = 1;
}
message StreamRegistryServiceDeleteRequest {
@@ -382,6 +384,7 @@ message MeasureRegistryServiceCreateRequest {
}
message MeasureRegistryServiceCreateResponse {
+ int64 mod_revision = 1;
}
message MeasureRegistryServiceUpdateRequest {
@@ -389,6 +392,7 @@ message MeasureRegistryServiceUpdateRequest {
}
message MeasureRegistryServiceUpdateResponse {
+ int64 mod_revision = 1;
}
message MeasureRegistryServiceDeleteRequest {
diff --git a/src/main/proto/banyandb/v1/banyandb-measure.proto b/src/main/proto/banyandb/v1/banyandb-measure.proto
index 1d89baa..574e68b 100644
--- a/src/main/proto/banyandb/v1/banyandb-measure.proto
+++ b/src/main/proto/banyandb/v1/banyandb-measure.proto
@@ -152,10 +152,19 @@ message WriteRequest {
common.v1.Metadata metadata = 1 [(validate.rules).message.required = true];
// the data_point is required.
DataPointValue data_point = 2 [(validate.rules).message.required = true];
+ // the message_id is required.
+ uint64 message_id = 3 [(validate.rules).uint64.gt = 0];
}
// WriteResponse is the response contract for write
-message WriteResponse {}
+message WriteResponse {
+ // the message_id from request.
+ uint64 message_id = 1 [(validate.rules).uint64.gt = 0];
+ // status indicates the request processing result
+ model.v1.Status status = 2 [(validate.rules).enum.defined_only = true];
+ // the metadata from request when request fails
+ common.v1.Metadata metadata = 3;
+}
service MeasureService {
rpc Query(banyandb.measure.v1.QueryRequest) returns (banyandb.measure.v1.QueryResponse);
diff --git a/src/main/proto/banyandb/v1/banyandb-model.proto b/src/main/proto/banyandb/v1/banyandb-model.proto
index 151fd0b..09824e3 100644
--- a/src/main/proto/banyandb/v1/banyandb-model.proto
+++ b/src/main/proto/banyandb/v1/banyandb-model.proto
@@ -177,3 +177,13 @@ message TimeRange {
google.protobuf.Timestamp begin = 1;
google.protobuf.Timestamp end = 2;
}
+
+// Status is the response status for write
+enum Status {
+ STATUS_UNSPECIFIED = 0;
+ STATUS_SUCCEED = 1;
+ STATUS_INVALID_TIMESTAMP = 2;
+ STATUS_NOT_FOUND = 3;
+ STATUS_EXPIRED_SCHEMA = 4;
+ STATUS_INTERNAL_ERROR = 5;
+}
\ No newline at end of file
diff --git a/src/main/proto/banyandb/v1/banyandb-stream.proto b/src/main/proto/banyandb/v1/banyandb-stream.proto
index 84338cf..29e73cf 100644
--- a/src/main/proto/banyandb/v1/banyandb-stream.proto
+++ b/src/main/proto/banyandb/v1/banyandb-stream.proto
@@ -84,13 +84,22 @@ message ElementValue {
}
message WriteRequest {
- // the metadata is only required in the first write.
+ // the metadata is required.
common.v1.Metadata metadata = 1 [(validate.rules).message.required = true];
// the element is required.
ElementValue element = 2 [(validate.rules).message.required = true];
+ // the message_id is required.
+ uint64 message_id = 3 [(validate.rules).uint64.gt = 0];
}
-message WriteResponse {}
+message WriteResponse {
+ // the message_id from request.
+ uint64 message_id = 1 [(validate.rules).uint64.gt = 0];
+ // status indicates the request processing result
+ model.v1.Status status = 2 [(validate.rules).enum.defined_only = true];
+ // the metadata from request when request fails
+ common.v1.Metadata metadata = 3;
+}
service StreamService {
rpc Query(banyandb.stream.v1.QueryRequest) returns (banyandb.stream.v1.QueryResponse);
diff --git a/src/test/java/org/apache/skywalking/banyandb/v1/client/AbstractBanyanDBClientTest.java b/src/test/java/org/apache/skywalking/banyandb/v1/client/AbstractBanyanDBClientTest.java
index a53e97e..328c0dc 100644
--- a/src/test/java/org/apache/skywalking/banyandb/v1/client/AbstractBanyanDBClientTest.java
+++ b/src/test/java/org/apache/skywalking/banyandb/v1/client/AbstractBanyanDBClientTest.java
@@ -40,6 +40,7 @@ import java.io.IOException;
import java.time.ZonedDateTime;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
import static org.mockito.AdditionalAnswers.delegatesTo;
import static org.mockito.Mockito.mock;
@@ -156,13 +157,18 @@ public class AbstractBanyanDBClientTest {
private final StreamRegistryServiceGrpc.StreamRegistryServiceImplBase streamRegistryServiceImpl =
mock(StreamRegistryServiceGrpc.StreamRegistryServiceImplBase.class, delegatesTo(
new StreamRegistryServiceGrpc.StreamRegistryServiceImplBase() {
+ private final AtomicLong revisionGenerator = new AtomicLong(0);
+
@Override
public void create(BanyandbDatabase.StreamRegistryServiceCreateRequest request, StreamObserver<BanyandbDatabase.StreamRegistryServiceCreateResponse> responseObserver) {
BanyandbDatabase.Stream s = request.getStream().toBuilder()
.setUpdatedAt(TimeUtils.buildTimestamp(ZonedDateTime.now()))
+ .setMetadata(request.getStream().getMetadata().toBuilder()
+ .setModRevision(revisionGenerator.incrementAndGet())
+ .build())
.build();
streamRegistry.put(s.getMetadata().getName(), s);
- responseObserver.onNext(BanyandbDatabase.StreamRegistryServiceCreateResponse.newBuilder().build());
+ responseObserver.onNext(BanyandbDatabase.StreamRegistryServiceCreateResponse.newBuilder().setModRevision(s.getMetadata().getModRevision()).build());
responseObserver.onCompleted();
}
@@ -170,9 +176,12 @@ public class AbstractBanyanDBClientTest {
public void update(BanyandbDatabase.StreamRegistryServiceUpdateRequest request, StreamObserver<BanyandbDatabase.StreamRegistryServiceUpdateResponse> responseObserver) {
BanyandbDatabase.Stream s = request.getStream().toBuilder()
.setUpdatedAt(TimeUtils.buildTimestamp(ZonedDateTime.now()))
+ .setMetadata(request.getStream().getMetadata().toBuilder()
+ .setModRevision(revisionGenerator.incrementAndGet())
+ .build())
.build();
streamRegistry.put(s.getMetadata().getName(), s);
- responseObserver.onNext(BanyandbDatabase.StreamRegistryServiceUpdateResponse.newBuilder().build());
+ responseObserver.onNext(BanyandbDatabase.StreamRegistryServiceUpdateResponse.newBuilder().setModRevision(s.getMetadata().getModRevision()).build());
responseObserver.onCompleted();
}
@@ -208,13 +217,18 @@ public class AbstractBanyanDBClientTest {
private final MeasureRegistryServiceGrpc.MeasureRegistryServiceImplBase measureRegistryServiceImpl =
mock(MeasureRegistryServiceGrpc.MeasureRegistryServiceImplBase.class, delegatesTo(
new MeasureRegistryServiceGrpc.MeasureRegistryServiceImplBase() {
+ private final AtomicLong revisionGenerator = new AtomicLong(0);
+
@Override
public void create(BanyandbDatabase.MeasureRegistryServiceCreateRequest request, StreamObserver<BanyandbDatabase.MeasureRegistryServiceCreateResponse> responseObserver) {
BanyandbDatabase.Measure s = request.getMeasure().toBuilder()
.setUpdatedAt(TimeUtils.buildTimestamp(ZonedDateTime.now()))
+ .setMetadata(request.getMeasure().getMetadata().toBuilder()
+ .setModRevision(revisionGenerator.incrementAndGet())
+ .build())
.build();
measureRegistry.put(s.getMetadata().getName(), s);
- responseObserver.onNext(BanyandbDatabase.MeasureRegistryServiceCreateResponse.newBuilder().build());
+ responseObserver.onNext(BanyandbDatabase.MeasureRegistryServiceCreateResponse.newBuilder().setModRevision(s.getMetadata().getModRevision()).build());
responseObserver.onCompleted();
}
@@ -222,9 +236,12 @@ public class AbstractBanyanDBClientTest {
public void update(BanyandbDatabase.MeasureRegistryServiceUpdateRequest request, StreamObserver<BanyandbDatabase.MeasureRegistryServiceUpdateResponse> responseObserver) {
BanyandbDatabase.Measure s = request.getMeasure().toBuilder()
.setUpdatedAt(TimeUtils.buildTimestamp(ZonedDateTime.now()))
+ .setMetadata(request.getMeasure().getMetadata().toBuilder()
+ .setModRevision(revisionGenerator.incrementAndGet())
+ .build())
.build();
measureRegistry.put(s.getMetadata().getName(), s);
- responseObserver.onNext(BanyandbDatabase.MeasureRegistryServiceUpdateResponse.newBuilder().build());
+ responseObserver.onNext(BanyandbDatabase.MeasureRegistryServiceUpdateResponse.newBuilder().setModRevision(s.getMetadata().getModRevision()).build());
responseObserver.onCompleted();
}
diff --git a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientMeasureWriteTest.java b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientMeasureWriteTest.java
index 7bb8437..f5b966a 100644
--- a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientMeasureWriteTest.java
+++ b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientMeasureWriteTest.java
@@ -18,16 +18,21 @@
package org.apache.skywalking.banyandb.v1.client;
+import io.grpc.ServerServiceDefinition;
import io.grpc.stub.StreamObserver;
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.measure.v1.BanyandbMeasure;
import org.apache.skywalking.banyandb.measure.v1.MeasureServiceGrpc;
+import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
import org.apache.skywalking.banyandb.v1.client.metadata.Duration;
import org.apache.skywalking.banyandb.v1.client.metadata.IndexRule;
import org.apache.skywalking.banyandb.v1.client.metadata.IndexRuleBinding;
import org.apache.skywalking.banyandb.v1.client.metadata.Measure;
+import org.apache.skywalking.banyandb.v1.client.metadata.MeasureMetadataRegistry;
import org.apache.skywalking.banyandb.v1.client.metadata.TagFamilySpec;
+
+import lombok.SneakyThrows;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -41,8 +46,11 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import static com.google.common.base.Preconditions.checkNotNull;
+
public class BanyanDBClientMeasureWriteTest extends AbstractBanyanDBClientTest {
private MeasureBulkWriteProcessor measureBulkWriteProcessor;
+ private Measure measure;
@Before
public void setUp() throws IOException, BanyanDBException {
@@ -51,7 +59,7 @@ public class BanyanDBClientMeasureWriteTest extends AbstractBanyanDBClientTest {
measureBulkWriteProcessor = client.buildMeasureWriteProcessor(1000, 1, 1);
- Measure m = Measure.create("sw_metric", "service_cpm_minute", Duration.ofHours(1))
+ measure = Measure.create("sw_metric", "service_cpm_minute", Duration.ofHours(1))
.setEntityRelativeTags("entity_id")
.addTagFamily(TagFamilySpec.create("default")
.addTagSpec(TagFamilySpec.TagSpec.newStringTag("entity_id"))
@@ -60,7 +68,7 @@ public class BanyanDBClientMeasureWriteTest extends AbstractBanyanDBClientTest {
.addField(Measure.FieldSpec.newIntField("value").compressWithZSTD().encodeWithGorilla().build())
.addIndex(IndexRule.create("scope", IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.SERIES))
.build();
- client.define(m);
+ client.define(measure);
}
@After
@@ -82,29 +90,35 @@ public class BanyanDBClientMeasureWriteTest extends AbstractBanyanDBClientTest {
final List<BanyandbMeasure.WriteRequest> writeRequestDelivered = new ArrayList<>();
// implement the fake service
- final MeasureServiceGrpc.MeasureServiceImplBase serviceImpl =
- new MeasureServiceGrpc.MeasureServiceImplBase() {
- @Override
- public StreamObserver<BanyandbMeasure.WriteRequest> write(StreamObserver<BanyandbMeasure.WriteResponse> responseObserver) {
- return new StreamObserver<BanyandbMeasure.WriteRequest>() {
- @Override
- public void onNext(BanyandbMeasure.WriteRequest value) {
- writeRequestDelivered.add(value);
- responseObserver.onNext(BanyandbMeasure.WriteResponse.newBuilder().build());
- }
-
- @Override
- public void onError(Throwable t) {
- }
-
- @Override
- public void onCompleted() {
- responseObserver.onCompleted();
- allRequestsDelivered.countDown();
- }
- };
- }
- };
+ final ServerServiceDefinition serviceImpl = createService(allRequestsDelivered, writeRequestDelivered);
+ serviceRegistry.addService(serviceImpl);
+
+ Instant now = Instant.now();
+ MeasureWrite measureWrite = client.createMeasureWrite("sw_metric", "service_cpm_minute", now.toEpochMilli());
+ measureWrite.tag("entity_id", TagAndValue.stringTagValue("entity_1"))
+ .field("total", TagAndValue.longFieldValue(100))
+ .field("value", TagAndValue.longFieldValue(1));
+
+ measureBulkWriteProcessor.add(measureWrite);
+
+ if (allRequestsDelivered.await(5, TimeUnit.SECONDS)) {
+ Assert.assertEquals(1, writeRequestDelivered.size());
+ final BanyandbMeasure.WriteRequest request = writeRequestDelivered.get(0);
+ Assert.assertEquals(1, request.getDataPoint().getTagFamilies(0).getTagsCount());
+ Assert.assertEquals("entity_1", request.getDataPoint().getTagFamilies(0).getTags(0).getStr().getValue());
+ Assert.assertEquals(100, request.getDataPoint().getFields(0).getInt().getValue());
+ Assert.assertEquals(1, request.getDataPoint().getFields(1).getInt().getValue());
+ Assert.assertTrue(request.getMessageId() > 0);
+ } else {
+ Assert.fail();
+ }
+ }
+
+ @Test
+ public void testAutoRefreshSchema() throws Exception {
+ CountDownLatch allRequestsDelivered = new CountDownLatch(1);
+ List<BanyandbMeasure.WriteRequest> writeRequestDelivered = new ArrayList<>();
+ ServerServiceDefinition serviceImpl = createService(allRequestsDelivered, writeRequestDelivered);
serviceRegistry.addService(serviceImpl);
Instant now = Instant.now();
@@ -113,8 +127,34 @@ public class BanyanDBClientMeasureWriteTest extends AbstractBanyanDBClientTest {
.field("total", TagAndValue.longFieldValue(100))
.field("value", TagAndValue.longFieldValue(1));
+ // update schema
+ Measure measureUpdate = measure.toBuilder()
+ .addField(Measure.FieldSpec.newIntField("new_field").compressWithZSTD().encodeWithGorilla().build())
+ .build();
+ MeasureMetadataRegistry measureMetadataRegistry = new MeasureMetadataRegistry(checkNotNull(this.channel));
+ measureMetadataRegistry.update(measureUpdate);
+
measureBulkWriteProcessor.add(measureWrite);
+ if (allRequestsDelivered.await(5, TimeUnit.SECONDS)) {
+ Assert.assertEquals(0, writeRequestDelivered.size());
+ } else {
+ Assert.fail();
+ }
+ // rewrite
+ serviceRegistry.removeService(serviceImpl);
+ allRequestsDelivered = new CountDownLatch(1);
+ writeRequestDelivered = new ArrayList<>();
+ serviceImpl = createService(allRequestsDelivered, writeRequestDelivered);
+ serviceRegistry.addService(serviceImpl);
+
+ now = Instant.now();
+ measureWrite = client.createMeasureWrite("sw_metric", "service_cpm_minute", now.toEpochMilli());
+ measureWrite.tag("entity_id", TagAndValue.stringTagValue("entity_1"))
+ .field("total", TagAndValue.longFieldValue(100))
+ .field("value", TagAndValue.longFieldValue(1));
+
+ measureBulkWriteProcessor.add(measureWrite);
if (allRequestsDelivered.await(5, TimeUnit.SECONDS)) {
Assert.assertEquals(1, writeRequestDelivered.size());
final BanyandbMeasure.WriteRequest request = writeRequestDelivered.get(0);
@@ -122,8 +162,50 @@ public class BanyanDBClientMeasureWriteTest extends AbstractBanyanDBClientTest {
Assert.assertEquals("entity_1", request.getDataPoint().getTagFamilies(0).getTags(0).getStr().getValue());
Assert.assertEquals(100, request.getDataPoint().getFields(0).getInt().getValue());
Assert.assertEquals(1, request.getDataPoint().getFields(1).getInt().getValue());
+ Assert.assertTrue(request.getMessageId() > 0);
} else {
Assert.fail();
}
}
+
+ @SneakyThrows
+ private boolean checkSchemaExpired(BanyandbMeasure.WriteRequest request) {
+ Measure m = client.findMeasure(request.getMetadata().getGroup(), request.getMetadata().getName());
+ return request.getMetadata().getModRevision() != m.serialize().getMetadata().getModRevision();
+ }
+
+ private ServerServiceDefinition createService(final CountDownLatch allRequestsDelivered,
+ final List<BanyandbMeasure.WriteRequest> writeRequestDelivered) {
+ return new MeasureServiceGrpc.MeasureServiceImplBase() {
+ @Override
+ public StreamObserver<BanyandbMeasure.WriteRequest> write(StreamObserver<BanyandbMeasure.WriteResponse> responseObserver) {
+ return new StreamObserver<BanyandbMeasure.WriteRequest>() {
+ @Override
+ public void onNext(BanyandbMeasure.WriteRequest request) {
+ if (checkSchemaExpired(request)) {
+ responseObserver.onNext(
+ BanyandbMeasure.WriteResponse.newBuilder()
+ .setMetadata(request.getMetadata())
+ .setStatus(BanyandbModel.Status.STATUS_EXPIRED_SCHEMA)
+ .setMessageId(request.getMessageId())
+ .build());
+ } else {
+ writeRequestDelivered.add(request);
+ responseObserver.onNext(BanyandbMeasure.WriteResponse.newBuilder().setMessageId(request.getMessageId()).build());
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ allRequestsDelivered.countDown();
+ }
+ };
+ }
+ }.bindService();
+ }
}
diff --git a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamWriteTest.java b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamWriteTest.java
index 2827987..6c741eb 100644
--- a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamWriteTest.java
+++ b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamWriteTest.java
@@ -19,17 +19,23 @@
package org.apache.skywalking.banyandb.v1.client;
import com.google.protobuf.NullValue;
+
+import io.grpc.ServerServiceDefinition;
import io.grpc.stub.StreamObserver;
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
import org.apache.skywalking.banyandb.database.v1.GroupRegistryServiceGrpc;
+import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
import org.apache.skywalking.banyandb.stream.v1.BanyandbStream;
import org.apache.skywalking.banyandb.stream.v1.StreamServiceGrpc;
import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
import org.apache.skywalking.banyandb.v1.client.metadata.IndexRule;
import org.apache.skywalking.banyandb.v1.client.metadata.IndexRuleBinding;
import org.apache.skywalking.banyandb.v1.client.metadata.Stream;
+import org.apache.skywalking.banyandb.v1.client.metadata.StreamMetadataRegistry;
import org.apache.skywalking.banyandb.v1.client.metadata.TagFamilySpec;
+
+import lombok.SneakyThrows;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -42,6 +48,7 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import static com.google.common.base.Preconditions.checkNotNull;
import static org.mockito.AdditionalAnswers.delegatesTo;
import static org.mockito.Mockito.mock;
@@ -65,13 +72,14 @@ public class BanyanDBClientStreamWriteTest extends AbstractBanyanDBClientTest {
}));
private StreamBulkWriteProcessor streamBulkWriteProcessor;
+ private Stream stream;
@Before
public void setUp() throws IOException, BanyanDBException {
setUp(bindService(groupRegistryServiceImpl), bindStreamRegistry());
streamBulkWriteProcessor = client.buildStreamWriteProcessor(1000, 1, 1);
- Stream expectedStream = Stream.create("default", "sw")
+ stream = Stream.create("default", "sw")
.setEntityRelativeTags("service_id", "service_instance_id", "state")
.addTagFamily(TagFamilySpec.create("data")
.addTagSpec(TagFamilySpec.TagSpec.newBinaryTag("data_binary"))
@@ -93,7 +101,7 @@ public class BanyanDBClientStreamWriteTest extends AbstractBanyanDBClientTest {
.build())
.addIndex(IndexRule.create("trace_id", IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.GLOBAL))
.build();
- this.client.define(expectedStream);
+ this.client.define(stream);
}
@After
@@ -194,29 +202,7 @@ public class BanyanDBClientStreamWriteTest extends AbstractBanyanDBClientTest {
final List<BanyandbStream.WriteRequest> writeRequestDelivered = new ArrayList<>();
// implement the fake service
- final StreamServiceGrpc.StreamServiceImplBase serviceImpl =
- new StreamServiceGrpc.StreamServiceImplBase() {
- @Override
- public StreamObserver<BanyandbStream.WriteRequest> write(StreamObserver<BanyandbStream.WriteResponse> responseObserver) {
- return new StreamObserver<BanyandbStream.WriteRequest>() {
- @Override
- public void onNext(BanyandbStream.WriteRequest value) {
- writeRequestDelivered.add(value);
- responseObserver.onNext(BanyandbStream.WriteResponse.newBuilder().build());
- }
-
- @Override
- public void onError(Throwable t) {
- }
-
- @Override
- public void onCompleted() {
- responseObserver.onCompleted();
- allRequestsDelivered.countDown();
- }
- };
- }
- };
+ final ServerServiceDefinition serviceImpl = createService(allRequestsDelivered, writeRequestDelivered);
serviceRegistry.addService(serviceImpl);
String segmentId = "1231.dfd.123123ssf";
@@ -262,8 +248,145 @@ public class BanyanDBClientStreamWriteTest extends AbstractBanyanDBClientTest {
Assert.assertEquals(latency, request.getElement().getTagFamilies(1).getTags(5).getInt().getValue());
Assert.assertEquals(request.getElement().getTagFamilies(1).getTags(7).getNull(), NullValue.NULL_VALUE);
Assert.assertEquals(queue, request.getElement().getTagFamilies(1).getTags(12).getStr().getValue());
+ Assert.assertTrue(request.getMessageId() > 0);
+ } else {
+ Assert.fail();
+ }
+ }
+
+ @Test
+ public void testAutoRefreshSchema() throws Exception {
+ CountDownLatch allRequestsDelivered = new CountDownLatch(1);
+ List<BanyandbStream.WriteRequest> writeRequestDelivered = new ArrayList<>();
+ ServerServiceDefinition serviceImpl = createService(allRequestsDelivered, writeRequestDelivered);
+ serviceRegistry.addService(serviceImpl);
+
+ String segmentId = "1231.dfd.123123ssf";
+ String traceId = "trace_id-xxfff.111323";
+ String serviceId = "webapp_id";
+ String serviceInstanceId = "10.0.0.1_id";
+ String endpointId = "home_id";
+ int latency = 200;
+ int state = 1;
+ Instant now = Instant.now();
+ byte[] byteData = new byte[] {14};
+ String broker = "172.16.10.129:9092";
+ String topic = "topic_1";
+ String queue = "queue_2";
+ String httpStatusCode = "200";
+ String dbType = "SQL";
+ String dbInstance = "127.0.0.1:3306";
+
+ 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("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("http.method", Value.stringTagValue(null)) // 6
+ .tag("status_code", Value.stringTagValue(httpStatusCode)) // 7
+ .tag("db.type", Value.stringTagValue(dbType)) // 8
+ .tag("db.instance", Value.stringTagValue(dbInstance)) // 9
+ .tag("mq.broker", Value.stringTagValue(broker)) // 10
+ .tag("mq.topic", Value.stringTagValue(topic)) // 11
+ .tag("mq.queue", Value.stringTagValue(queue)); // 12
+
+ // update schema
+ Stream streamUpdate = stream.toBuilder()
+ .addTagFamily(TagFamilySpec.create("new_tag_family")
+ .addTagSpec(TagFamilySpec.TagSpec.newIntTag("new_tag"))
+ .build())
+ .build();
+ StreamMetadataRegistry streamMetadataRegistry = new StreamMetadataRegistry(checkNotNull(this.channel));
+ streamMetadataRegistry.update(streamUpdate);
+
+ streamBulkWriteProcessor.add(streamWrite);
+ if (allRequestsDelivered.await(5, TimeUnit.SECONDS)) {
+ Assert.assertEquals(0, writeRequestDelivered.size());
+ } else {
+ Assert.fail();
+ }
+
+ // rewrite
+ serviceRegistry.removeService(serviceImpl);
+ allRequestsDelivered = new CountDownLatch(1);
+ writeRequestDelivered = new ArrayList<>();
+ serviceImpl = createService(allRequestsDelivered, writeRequestDelivered);
+ serviceRegistry.addService(serviceImpl);
+
+ now = Instant.now();
+ 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("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("http.method", Value.stringTagValue(null)) // 6
+ .tag("status_code", Value.stringTagValue(httpStatusCode)) // 7
+ .tag("db.type", Value.stringTagValue(dbType)) // 8
+ .tag("db.instance", Value.stringTagValue(dbInstance)) // 9
+ .tag("mq.broker", Value.stringTagValue(broker)) // 10
+ .tag("mq.topic", Value.stringTagValue(topic)) // 11
+ .tag("mq.queue", Value.stringTagValue(queue)); // 12
+
+ streamBulkWriteProcessor.add(streamWrite);
+ if (allRequestsDelivered.await(5, TimeUnit.SECONDS)) {
+ Assert.assertEquals(1, writeRequestDelivered.size());
+ final BanyandbStream.WriteRequest request = writeRequestDelivered.get(0);
+ Assert.assertArrayEquals(byteData, request.getElement().getTagFamilies(0).getTags(0).getBinaryData().toByteArray());
+ Assert.assertEquals(13, request.getElement().getTagFamilies(1).getTagsCount());
+ Assert.assertEquals(traceId, request.getElement().getTagFamilies(1).getTags(0).getStr().getValue());
+ Assert.assertEquals(latency, request.getElement().getTagFamilies(1).getTags(5).getInt().getValue());
+ Assert.assertEquals(request.getElement().getTagFamilies(1).getTags(7).getNull(), NullValue.NULL_VALUE);
+ Assert.assertEquals(queue, request.getElement().getTagFamilies(1).getTags(12).getStr().getValue());
+ Assert.assertTrue(request.getMessageId() > 0);
} else {
Assert.fail();
}
}
+
+ @SneakyThrows
+ private boolean checkSchemaExpired(BanyandbStream.WriteRequest request) {
+ Stream m = client.findStream(request.getMetadata().getGroup(), request.getMetadata().getName());
+ return request.getMetadata().getModRevision() != m.serialize().getMetadata().getModRevision();
+ }
+
+ private ServerServiceDefinition createService(final CountDownLatch allRequestsDelivered,
+ final List<BanyandbStream.WriteRequest> writeRequestDelivered) {
+ return new StreamServiceGrpc.StreamServiceImplBase() {
+ @Override
+ public StreamObserver<BanyandbStream.WriteRequest> write(StreamObserver<BanyandbStream.WriteResponse> responseObserver) {
+ return new StreamObserver<BanyandbStream.WriteRequest>() {
+ @Override
+ public void onNext(BanyandbStream.WriteRequest request) {
+ if (checkSchemaExpired(request)) {
+ responseObserver.onNext(
+ BanyandbStream.WriteResponse.newBuilder()
+ .setMetadata(request.getMetadata())
+ .setStatus(BanyandbModel.Status.STATUS_EXPIRED_SCHEMA)
+ .setMessageId(request.getMessageId())
+ .build());
+ } else {
+ writeRequestDelivered.add(request);
+ responseObserver.onNext(BanyandbStream.WriteResponse.newBuilder().setMessageId(request.getMessageId()).build());
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ allRequestsDelivered.countDown();
+ }
+ };
+ }
+ }.bindService();
+ }
}
diff --git a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java
index d91f987..b107273 100644
--- a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java
+++ b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java
@@ -30,7 +30,7 @@ import java.io.IOException;
public class BanyanDBClientTestCI {
private static final String REGISTRY = "ghcr.io";
private static final String IMAGE_NAME = "apache/skywalking-banyandb";
- private static final String TAG = "adbd3e87df7f84e5d1904fcf40476d2e81842058";
+ private static final String TAG = "9552688d60f055f6bb539a91464e8b1db77e7309";
private static final String IMAGE = REGISTRY + "/" + IMAGE_NAME + ":" + TAG;
diff --git a/src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/MeasureMetadataRegistryTest.java b/src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/MeasureMetadataRegistryTest.java
index 159c1df..e87bef1 100644
--- a/src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/MeasureMetadataRegistryTest.java
+++ b/src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/MeasureMetadataRegistryTest.java
@@ -52,6 +52,7 @@ public class MeasureMetadataRegistryTest extends AbstractBanyanDBClientTest {
Assert.assertNotNull(actualMeasure);
Assert.assertEquals(expectedMeasure, actualMeasure);
Assert.assertNotNull(actualMeasure.updatedAt());
+ Assert.assertNotNull(actualMeasure.modRevision());
}
@Test
@@ -71,6 +72,7 @@ public class MeasureMetadataRegistryTest extends AbstractBanyanDBClientTest {
List<Measure> actualMeasures = new MeasureMetadataRegistry(this.channel).list("sw_metric");
Assert.assertNotNull(actualMeasures);
Assert.assertEquals(1, actualMeasures.size());
+ actualMeasures.forEach(measure -> Assert.assertNotNull(measure.modRevision()));
}
@Test
@@ -87,7 +89,7 @@ public class MeasureMetadataRegistryTest extends AbstractBanyanDBClientTest {
.addIndex(IndexRule.create("scope", IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.SERIES))
.build();
this.client.define(expectedMeasure);
- boolean deleted = new MeasureMetadataRegistry(this.channel).delete("sw_metric", "service_cpm_minute");
+ boolean deleted = this.client.delete(expectedMeasure);
Assert.assertTrue(deleted);
Assert.assertEquals(0, measureRegistry.size());
}
diff --git a/src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/StreamMetadataRegistryTest.java b/src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/StreamMetadataRegistryTest.java
index 5007965..140608a 100644
--- a/src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/StreamMetadataRegistryTest.java
+++ b/src/test/java/org/apache/skywalking/banyandb/v1/client/metadata/StreamMetadataRegistryTest.java
@@ -54,6 +54,7 @@ public class StreamMetadataRegistryTest extends AbstractBanyanDBClientTest {
Assert.assertNotNull(actualStream);
Assert.assertEquals(expectedStream, actualStream);
Assert.assertNotNull(actualStream.updatedAt());
+ Assert.assertNotNull(actualStream.modRevision());
}
@Test
@@ -74,6 +75,7 @@ public class StreamMetadataRegistryTest extends AbstractBanyanDBClientTest {
List<Stream> actualStreams = new StreamMetadataRegistry(this.channel).list("default");
Assert.assertNotNull(actualStreams);
Assert.assertEquals(1, actualStreams.size());
+ actualStreams.forEach(stream -> Assert.assertNotNull(stream.modRevision()));
}
@Test
@@ -91,7 +93,7 @@ public class StreamMetadataRegistryTest extends AbstractBanyanDBClientTest {
.addIndex(IndexRule.create("trace_id", IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.GLOBAL))
.build();
this.client.define(expectedStream);
- boolean deleted = new StreamMetadataRegistry(this.channel).delete("default", "sw");
+ boolean deleted = this.client.delete(expectedStream);
Assert.assertTrue(deleted);
Assert.assertEquals(0, streamRegistry.size());
}