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

Author: hailin0

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
 ------------------
 

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,15 +280,15 @@ 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,
 
 ```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));

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

@@ -74,7 +74,7 @@ P build() {
         }
 
         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();

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

@@ -28,6 +28,8 @@
 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.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.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,33 +203,73 @@ void connect(Channel channel) {
      * 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 {
             writeRequestStreamObserver.onNext(streamWrite.build());
         } finally {
             writeRequestStreamObserver.onCompleted();
         }
+        return future;
     }
 
     /**
@@ -239,7 +284,7 @@ public void onCompleted() {
     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 StreamBulkWriteProcessor buildStreamWriteProcessor(int maxBulkSize, int f
     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,23 +406,57 @@ public Group define(Group group) throws BanyanDBException {
      */
     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
      *
      * @param measure the measure to be created
      */
     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
      *

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

@@ -20,10 +20,16 @@
 
 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 @@
 @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

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

@@ -93,6 +93,7 @@ protected BanyandbMeasure.WriteRequest build(BanyandbCommon.Metadata metadata, T
         }
 
         builder.setDataPoint(datapointValueBuilder);
+        builder.setMessageId(System.nanoTime());
         return builder.build();
     }
 }

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

@@ -21,10 +21,16 @@
 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 @@
 @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

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

@@ -94,6 +94,7 @@ protected BanyandbStream.WriteRequest build(BanyandbCommon.Metadata metadata, Ti
             elemValBuilder.addTagFamilies(BanyandbModel.TagFamilyForWrite.newBuilder().addAllTags(tags).build());
         }
         builder.setElement(elemValBuilder);
+        builder.setMessageId(System.nanoTime());
         return builder.build();
     }
 }