feat: Support shards stage1 (#72)

Author: s12f

.github/workflows/main.yml

@@ -34,7 +34,7 @@ jobs:
         run: ./gradlew ktlintCheck
 
       - name: Prepare test env
-        run: script/prepare-test-env.sh
+        run:  script/prepare-test-env-latest.sh
 
       - name: Build and test with Gradle
         run: ./gradlew build

client/src/main/java/io/hstream/Producer.java

@@ -20,4 +20,12 @@ public interface Producer {
    * @return the {@link RecordId} wrapped in a {@link CompletableFuture}
    */
   CompletableFuture<RecordId> write(HRecord hRecord);
+
+  /**
+   * Write a {@link Record}.
+   *
+   * @param record {@link Record}
+   * @return the {@link RecordId} wrapped in a {@link CompletableFuture}
+   */
+  CompletableFuture<RecordId> write(Record record);
 }

client/src/main/java/io/hstream/Record.java

@@ -0,0 +1,76 @@
+package io.hstream;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+public class Record {
+
+  private String key;
+  private byte[] rawRecord;
+  private HRecord hRecord;
+  private boolean isRawRecord;
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  private Record(String key, byte[] rawRecord) {
+    isRawRecord = true;
+    this.key = key;
+    this.rawRecord = rawRecord;
+  }
+
+  private Record(String key, HRecord hRecord) {
+    isRawRecord = false;
+    this.key = key;
+    this.hRecord = hRecord;
+  }
+
+  public String getKey() {
+    return key;
+  }
+
+  public byte[] getRawRecord() {
+    checkArgument(isRawRecord);
+    return rawRecord;
+  }
+
+  public HRecord getHRecord() {
+    checkArgument(!isRawRecord);
+    return hRecord;
+  }
+
+  public boolean isRawRecord() {
+    return isRawRecord;
+  }
+
+  public static class Builder {
+    private String key;
+    private byte[] rawRecord;
+    private HRecord hRecord;
+
+    public Builder key(String key) {
+      this.key = key;
+      return this;
+    }
+
+    public Builder rawRecord(byte[] rawRecord) {
+      this.rawRecord = rawRecord;
+      return this;
+    }
+
+    public Builder hRecord(HRecord hRecord) {
+      this.hRecord = hRecord;
+      return this;
+    }
+
+    public Record build() {
+      checkArgument(
+          (rawRecord != null && hRecord == null) || (rawRecord == null && hRecord != null));
+      if (rawRecord != null) {
+        return new Record(key, rawRecord);
+      } else {
+        return new Record(key, hRecord);
+      }
+    }
+  }
+}

client/src/main/java/io/hstream/Subscription.java

@@ -8,17 +8,11 @@ public class Subscription {
 
   private String subscriptionId;
   private String streamName;
-  private SubscriptionOffset subscriptionOffset;
   private int ackTimeoutSeconds;
 
-  private Subscription(
-      String subscriptionId,
-      String streamName,
-      SubscriptionOffset subscriptionOffset,
-      int ackTimeoutSeconds) {
+  private Subscription(String subscriptionId, String streamName, int ackTimeoutSeconds) {
     this.subscriptionId = subscriptionId;
     this.streamName = streamName;
-    this.subscriptionOffset = subscriptionOffset;
     this.ackTimeoutSeconds = ackTimeoutSeconds;
   }
 
@@ -35,10 +29,6 @@ public String getStreamName() {
     return streamName;
   }
 
-  public SubscriptionOffset getSubscriptionOffset() {
-    return subscriptionOffset;
-  }
-
   public int getAckTimeoutSeconds() {
     return ackTimeoutSeconds;
   }
@@ -50,20 +40,18 @@ public boolean equals(Object o) {
     Subscription that = (Subscription) o;
     return ackTimeoutSeconds == that.ackTimeoutSeconds
         && subscriptionId.equals(that.subscriptionId)
-        && streamName.equals(that.streamName)
-        && subscriptionOffset.equals(that.subscriptionOffset);
+        && streamName.equals(that.streamName);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(subscriptionId, streamName, subscriptionOffset, ackTimeoutSeconds);
+    return Objects.hash(subscriptionId, streamName, ackTimeoutSeconds);
   }
 
   public static class Builder {
 
     private String subscriptionId;
     private String streamName;
-    private SubscriptionOffset subscriptionOffset;
     private int ackTimeoutSeconds = 600;
 
     public Builder subscription(String subscriptionId) {
@@ -76,11 +64,6 @@ public Builder stream(String streamName) {
       return this;
     }
 
-    public Builder offset(SubscriptionOffset subscriptionOffset) {
-      this.subscriptionOffset = subscriptionOffset;
-      return this;
-    }
-
     public Builder ackTimeoutSeconds(int ackTimeoutSeconds) {
       this.ackTimeoutSeconds = ackTimeoutSeconds;
       return this;
@@ -89,9 +72,8 @@ public Builder ackTimeoutSeconds(int ackTimeoutSeconds) {
     public Subscription build() {
       checkNotNull(subscriptionId);
       checkNotNull(streamName);
-      checkNotNull(subscriptionOffset);
       checkState(ackTimeoutSeconds > 0 && ackTimeoutSeconds < 36000);
-      return new Subscription(subscriptionId, streamName, subscriptionOffset, ackTimeoutSeconds);
+      return new Subscription(subscriptionId, streamName, ackTimeoutSeconds);
     }
   }
 }

client/src/main/java/io/hstream/SubscriptionOffset.java

@@ -5,4 +5,6 @@ public class DefaultSettings {
   public static final long REQUEST_RETRY_INTERVAL_SECONDS = 5;
 
   public static final int APPEND_RETRY_MAX_TIMES = 5;
+
+  public static final String DEFAULT_ORDERING_KEY = "__default__";
 }

client/src/main/java/io/hstream/impl/DefaultSettings.java

@@ -4,7 +4,6 @@
 import io.grpc.stub.StreamObserver;
 import io.hstream.*;
 import io.hstream.Subscription;
-import io.hstream.SubscriptionOffset;
 import io.hstream.internal.*;
 import io.hstream.internal.Stream;
 import java.util.List;
@@ -76,7 +75,6 @@ public void onNext(CreateQueryStreamResponse value) {
                 Subscription.newBuilder()
                     .subscription(STREAM_QUERY_SUBSCRIPTION_PREFIX + resultStreamNameSuffix)
                     .stream(STREAM_QUERY_STREAM_PREFIX + resultStreamNameSuffix)
-                    .offset(new SubscriptionOffset(SubscriptionOffset.SpecialOffset.EARLIEST))
                     .ackTimeoutSeconds(10)
                     .build();
             client.createSubscription(subscription);

client/src/main/java/io/hstream/impl/QueryerImpl.java

@@ -19,66 +19,17 @@ public static RecordId recordIdFromGrpc(io.hstream.internal.RecordId recordId) {
     return new RecordId(recordId.getBatchId(), recordId.getBatchIndex());
   }
 
-  public static io.hstream.internal.SubscriptionOffset.SpecialOffset specialOffsetToGrpc(
-      SubscriptionOffset.SpecialOffset offset) {
-    switch (offset) {
-      case EARLIEST:
-        return io.hstream.internal.SubscriptionOffset.SpecialOffset.EARLIST;
-      case LATEST:
-        return io.hstream.internal.SubscriptionOffset.SpecialOffset.LATEST;
-      default:
-        throw new IllegalArgumentException();
-    }
-  }
-
-  public static SubscriptionOffset.SpecialOffset specialOffsetFromGrpc(
-      io.hstream.internal.SubscriptionOffset.SpecialOffset offset) {
-    switch (offset) {
-      case EARLIST:
-        return SubscriptionOffset.SpecialOffset.EARLIEST;
-      case LATEST:
-        return SubscriptionOffset.SpecialOffset.LATEST;
-      default:
-        throw new IllegalArgumentException();
-    }
-  }
-
-  public static io.hstream.internal.SubscriptionOffset subscriptionOffsetToGrpc(
-      SubscriptionOffset offset) {
-    if (offset.isSpecialOffset()) {
-      return io.hstream.internal.SubscriptionOffset.newBuilder()
-          .setSpecialOffset(specialOffsetToGrpc(offset.getSpecialOffset()))
-          .build();
-
-    } else {
-      return io.hstream.internal.SubscriptionOffset.newBuilder()
-          .setRecordOffset(recordIdToGrpc(offset.getNormalOffset()))
-          .build();
-    }
-  }
-
-  public static SubscriptionOffset subscriptionOffsetFromGrpc(
-      io.hstream.internal.SubscriptionOffset offset) {
-    if (offset.hasRecordOffset()) {
-      return new SubscriptionOffset(recordIdFromGrpc(offset.getRecordOffset()));
-    } else {
-      return new SubscriptionOffset(specialOffsetFromGrpc(offset.getSpecialOffset()));
-    }
-  }
-
   public static io.hstream.internal.Subscription subscriptionToGrpc(Subscription subscription) {
     return io.hstream.internal.Subscription.newBuilder()
         .setSubscriptionId(subscription.getSubscriptionId())
         .setStreamName(subscription.getStreamName())
-        .setOffset(subscriptionOffsetToGrpc(subscription.getSubscriptionOffset()))
         .setAckTimeoutSeconds(subscription.getAckTimeoutSeconds())
         .build();
   }
 
   public static Subscription subscriptionFromGrpc(io.hstream.internal.Subscription subscription) {
     return Subscription.newBuilder().subscription(subscription.getSubscriptionId()).stream(
             subscription.getStreamName())
-        .offset(subscriptionOffsetFromGrpc(subscription.getOffset()))
         .ackTimeoutSeconds(subscription.getAckTimeoutSeconds())
         .build();
   }

client/src/main/java/io/hstream/util/GrpcUtils.java

@@ -5,6 +5,8 @@
 import com.google.protobuf.Struct;
 import com.google.protobuf.util.JsonFormat;
 import io.hstream.*;
+import io.hstream.Record;
+import io.hstream.impl.DefaultSettings;
 import io.hstream.internal.HStreamRecord;
 import io.hstream.internal.HStreamRecordHeader;
 import io.hstream.internal.ReceivedRecord;
@@ -18,7 +20,10 @@ public class RecordUtils {
 
   public static HStreamRecord buildHStreamRecordFromRawRecord(byte[] rawRecord) {
     HStreamRecordHeader header =
-        HStreamRecordHeader.newBuilder().setFlag(HStreamRecordHeader.Flag.RAW).build();
+        HStreamRecordHeader.newBuilder()
+            .setFlag(HStreamRecordHeader.Flag.RAW)
+            .setKey(DefaultSettings.DEFAULT_ORDERING_KEY)
+            .build();
     return HStreamRecord.newBuilder()
         .setHeader(header)
         .setPayload(ByteString.copyFrom(rawRecord))
@@ -28,7 +33,10 @@ public static HStreamRecord buildHStreamRecordFromRawRecord(byte[] rawRecord) {
   public static HStreamRecord buildHStreamRecordFromHRecord(HRecord hRecord) {
     try {
       HStreamRecordHeader header =
-          HStreamRecordHeader.newBuilder().setFlag(HStreamRecordHeader.Flag.JSON).build();
+          HStreamRecordHeader.newBuilder()
+              .setFlag(HStreamRecordHeader.Flag.JSON)
+              .setKey(DefaultSettings.DEFAULT_ORDERING_KEY)
+              .build();
       String json = JsonFormat.printer().print(hRecord.getDelegate());
       logger.debug("hrecord to json: {}", json);
       return HStreamRecord.newBuilder()
@@ -40,6 +48,19 @@ public static HStreamRecord buildHStreamRecordFromHRecord(HRecord hRecord) {
     }
   }
 
+  public static HStreamRecord buildHStreamRecordFromRecord(Record record) {
+    HStreamRecord hStreamRecord =
+        record.isRawRecord()
+            ? buildHStreamRecordFromRawRecord(record.getRawRecord())
+            : buildHStreamRecordFromHRecord(record.getHRecord());
+    if (record.getKey() == null) {
+      return hStreamRecord;
+    }
+    HStreamRecordHeader newHeader =
+        HStreamRecordHeader.newBuilder(hStreamRecord.getHeader()).setKey(record.getKey()).build();
+    return HStreamRecord.newBuilder(hStreamRecord).setHeader(newHeader).build();
+  }
+
   public static byte[] parseRawRecordFromHStreamRecord(HStreamRecord hStreamRecord) {
     HStreamRecordHeader.Flag flag = hStreamRecord.getHeader().getFlag();
     if (!flag.equals(HStreamRecordHeader.Flag.RAW)) {