fix: update Record for ordering key (#76)

Author: s12f

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

@@ -2,7 +2,7 @@
 
 import java.io.Closeable;
 
-/** The interface for the HStream BuffetedProducer. */
+/** The interface for the HStream BufferedProducer. */
 public interface BufferedProducer extends Producer, Closeable {
 
   /** explicitly flush buffered records. */

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

@@ -5,12 +5,24 @@ public interface BufferedProducerBuilder {
 
   BufferedProducerBuilder stream(String streamName);
 
+  /** @param recordCountLimit optional, default: 100, it MUST be greater than 0 */
   BufferedProducerBuilder recordCountLimit(int recordCountLimit);
 
+  /**
+   * @param flushIntervalMs optional, default: 100(ms), disabled if flushIntervalMs {@literal <=} 0
+   */
   BufferedProducerBuilder flushIntervalMs(long flushIntervalMs);
 
+  /**
+   * @param maxBytesSize optional, default: 4096(Bytes), disabled if maxBytesSize {@literal <=} 0
+   */
   BufferedProducerBuilder maxBytesSize(int maxBytesSize);
 
+  /**
+   * @param throwExceptionIfFull optional, default: false, if throwExceptionIfFull is true, throw
+   *     HStreamDBClientException when buffer is full, otherwise, block thread and wait the buffer
+   *     to be flushed.
+   */
   BufferedProducerBuilder throwExceptionIfFull(boolean throwExceptionIfFull);
 
   BufferedProducer build();

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

@@ -11,6 +11,7 @@ public interface Producer {
    * @param rawRecord raw format record
    * @return the {@link RecordId} wrapped in a {@link CompletableFuture}
    */
+  @Deprecated
   CompletableFuture<RecordId> write(byte[] rawRecord);
 
   /**
@@ -19,6 +20,7 @@ public interface Producer {
    * @param hRecord {@link HRecord}
    * @return the {@link RecordId} wrapped in a {@link CompletableFuture}
    */
+  @Deprecated
   CompletableFuture<RecordId> write(HRecord hRecord);
 
   /**

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

@@ -4,7 +4,7 @@
 
 public class Record {
 
-  private String key;
+  private String orderingKey;
   private byte[] rawRecord;
   private HRecord hRecord;
   private boolean isRawRecord;
@@ -13,20 +13,20 @@ public static Builder newBuilder() {
     return new Builder();
   }
 
-  private Record(String key, byte[] rawRecord) {
+  private Record(String orderingKey, byte[] rawRecord) {
     isRawRecord = true;
-    this.key = key;
+    this.orderingKey = orderingKey;
     this.rawRecord = rawRecord;
   }
 
-  private Record(String key, HRecord hRecord) {
+  private Record(String orderingKey, HRecord hRecord) {
     isRawRecord = false;
-    this.key = key;
+    this.orderingKey = orderingKey;
     this.hRecord = hRecord;
   }
 
-  public String getKey() {
-    return key;
+  public String getOrderingKey() {
+    return orderingKey;
   }
 
   public byte[] getRawRecord() {
@@ -44,12 +44,12 @@ public boolean isRawRecord() {
   }
 
   public static class Builder {
-    private String key;
+    private String orderingKey;
     private byte[] rawRecord;
     private HRecord hRecord;
 
-    public Builder key(String key) {
-      this.key = key;
+    public Builder orderingKey(String key) {
+      this.orderingKey = key;
       return this;
     }
 
@@ -67,9 +67,9 @@ public Record build() {
       checkArgument(
           (rawRecord != null && hRecord == null) || (rawRecord == null && hRecord != null));
       if (rawRecord != null) {
-        return new Record(key, rawRecord);
+        return new Record(orderingKey, rawRecord);
       } else {
-        return new Record(key, hRecord);
+        return new Record(orderingKey, hRecord);
       }
     }
   }

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

@@ -53,11 +53,13 @@ public static HStreamRecord buildHStreamRecordFromRecord(Record record) {
         record.isRawRecord()
             ? buildHStreamRecordFromRawRecord(record.getRawRecord())
             : buildHStreamRecordFromHRecord(record.getHRecord());
-    if (record.getKey() == null) {
+    if (record.getOrderingKey() == null) {
       return hStreamRecord;
     }
     HStreamRecordHeader newHeader =
-        HStreamRecordHeader.newBuilder(hStreamRecord.getHeader()).setKey(record.getKey()).build();
+        HStreamRecordHeader.newBuilder(hStreamRecord.getHeader())
+            .setKey(record.getOrderingKey())
+            .build();
     return HStreamRecord.newBuilder(hStreamRecord).setHeader(newHeader).build();
   }
 

client/src/test/java/io/hstream/HStreamClientTest07.java

@@ -45,7 +45,7 @@ void writeTest() throws Exception {
     random.nextBytes(payload);
     Producer producer = client.newProducer().stream(streamName).build();
     CompletableFuture<RecordId> future =
-        producer.write(Record.newBuilder().key("k1").rawRecord(payload).build());
+        producer.write(Record.newBuilder().orderingKey("k1").rawRecord(payload).build());
     RecordId recordId = future.join();
     logger.info("write successfully, got recordId: " + recordId.toString());
     client.close();
@@ -72,7 +72,7 @@ void readTest() throws Exception {
           producer
               .write(
                   Record.newBuilder()
-                      .key("k1")
+                      .orderingKey("k1")
                       .rawRecord(("record-" + i).getBytes(StandardCharsets.UTF_8))
                       .build())
               .join();
@@ -117,7 +117,7 @@ void readMultipleShardsTest() throws Exception {
             producer
                 .write(
                     Record.newBuilder()
-                        .key(orderingKey)
+                        .orderingKey(orderingKey)
                         .rawRecord(("record-" + i).getBytes(StandardCharsets.UTF_8))
                         .build())
                 .join();
@@ -162,7 +162,7 @@ void consumerRebalenceTest() throws Exception {
             producer
                 .write(
                     Record.newBuilder()
-                        .key(orderingKey)
+                        .orderingKey(orderingKey)
                         .rawRecord(("record-" + i).getBytes(StandardCharsets.UTF_8))
                         .build())
                 .join();
@@ -231,7 +231,7 @@ void consumerRebalenceTest() throws Exception {
             producer
                 .write(
                     Record.newBuilder()
-                        .key(orderingKey)
+                        .orderingKey(orderingKey)
                         .rawRecord(("record-" + i).getBytes(StandardCharsets.UTF_8))
                         .build())
                 .join();

client/src/test/java/io/hstream/TestUtils.java

@@ -57,7 +57,7 @@ public static ArrayList<String> doProduce(
     var xs = new CompletableFuture[recordsNums];
     for (int i = 0; i < recordsNums; i++) {
       rand.nextBytes(rRec);
-      Record recordToWrite = Record.newBuilder().key(key).rawRecord(rRec).build();
+      Record recordToWrite = Record.newBuilder().orderingKey(key).rawRecord(rRec).build();
       records.add(Arrays.toString(rRec));
       xs[i] = producer.write(recordToWrite);
     }
@@ -68,7 +68,7 @@ public static ArrayList<String> doProduce(
   public static RecordId produceIntegerAndGatherRid(Producer producer, int data, String key) {
     Record recordToWrite =
         Record.newBuilder()
-            .key(key)
+            .orderingKey(key)
             .rawRecord(Integer.toString(data).getBytes(StandardCharsets.UTF_8))
             .build();
     return producer.write(recordToWrite).join();