updates

Author: sgbalogh

README.md

@@ -99,7 +99,7 @@ For the demos discussed below, it will be helpful to create a new basin and stre
 Start by setting some environment variables in your shell.
 
 ```bash
-export S2_AUTH_TOKEN="MY-SECRET"
+export S2_ACCESS_TOKEN="MY-SECRET"
 export S2_BASIN="my-demo-java"
 export S2_STREAM="test/1"
 ```

app/src/main/java/org/example/app/AccountDemo.java

@@ -8,18 +8,21 @@
 import s2.config.Config;
 import s2.config.Endpoints;
 import s2.types.Age;
+import s2.types.BasinConfig;
 import s2.types.CreateBasinRequest;
 import s2.types.ListBasinsRequest;
 import s2.types.StorageClass;
 import s2.types.StreamConfig;
+import s2.types.Timestamping;
+import s2.types.TimestampingMode;
 
 public class AccountDemo {
 
   private static final Logger logger = LoggerFactory.getLogger(AccountDemo.class.getName());
 
   public static void main(String[] args) throws Exception {
     var config =
-        Config.newBuilder(System.getenv("S2_AUTH_TOKEN"))
+        Config.newBuilder(System.getenv("S2_ACCESS_TOKEN"))
             .withEndpoints(Endpoints.fromEnvironment())
             .build();
 
@@ -33,11 +36,16 @@ public static void main(String[] args) throws Exception {
               .createBasin(
                   CreateBasinRequest.newBuilder()
                       .withBasin(UUID.randomUUID().toString())
-                      .withDefaultStreamConfig(
-                          StreamConfig.newBuilder()
-                              .withRetentionPolicy(new Age(Duration.ofDays(7)))
-                              .withStorageClass(StorageClass.STANDARD)
-                              .build())
+                      .withBasinConfig(
+                          new BasinConfig(
+                              StreamConfig.newBuilder()
+                                  .withRetentionPolicy(new Age(Duration.ofDays(7)))
+                                  .withTimestamping(
+                                      new Timestamping(TimestampingMode.CLIENT_REQUIRE, true))
+                                  .withStorageClass(StorageClass.STANDARD)
+                                  .build(),
+                              false,
+                              false))
                       .build())
               .get();
       logger.info("newBasin={}", newBasin);

app/src/main/java/org/example/app/BasinDemo.java

@@ -15,7 +15,7 @@ public class BasinDemo {
 
   public static void main(String[] args) throws Exception {
     final var config =
-        Config.newBuilder(System.getenv("S2_AUTH_TOKEN"))
+        Config.newBuilder(System.getenv("S2_ACCESS_TOKEN"))
             .withEndpoints(Endpoints.fromEnvironment())
             .build();
 

app/src/main/java/org/example/app/ManagedAppendSessionDemo.java

@@ -25,12 +25,15 @@ public class ManagedAppendSessionDemo {
   private static final Logger logger =
       LoggerFactory.getLogger(ManagedAppendSessionDemo.class.getName());
 
+  // 512KiB
+  private static final Integer TARGET_BATCH_SIZE = 512 * 1024;
+
   public static void main(String[] args) throws Exception {
-    final var authToken = System.getenv("S2_AUTH_TOKEN");
+    final var authToken = System.getenv("S2_ACCESS_TOKEN");
     final var basinName = System.getenv("S2_BASIN");
     final var streamName = System.getenv("S2_STREAM");
     if (authToken == null) {
-      throw new IllegalStateException("S2_AUTH_TOKEN not set");
+      throw new IllegalStateException("S2_ACCESS_TOKEN not set");
     }
     if (basinName == null) {
       throw new IllegalStateException("S2_BASIN not set");
@@ -82,7 +85,16 @@ public static void main(String[] args) throws Exception {
           try {
             // Generate a record with approximately 10KiB of random text.
             var payload =
-                RandomASCIIStringGenerator.generateRandomASCIIString(i + " - ", 1024 * 10);
+                RandomASCIIStringGenerator.generateRandomASCIIString(i + " - ", TARGET_BATCH_SIZE);
+
+            while (futureAppendSession.remainingBufferCapacityBytes()
+                < (TARGET_BATCH_SIZE + payload.length())) {
+              // Crude backpressure mechanism; slow down the rate of payload creation by sleeping
+              // momentarily
+              // if we have hit the internal append buffer max size.
+              Thread.sleep(10);
+            }
+
             var append =
                 futureAppendSession.submit(
                     AppendInput.newBuilder()

app/src/main/java/org/example/app/ManagedReadSessionDemo.java

@@ -11,6 +11,7 @@
 import s2.config.Endpoints;
 import s2.types.Batch;
 import s2.types.ReadSessionRequest;
+import s2.types.Start;
 
 public class ManagedReadSessionDemo {
 
@@ -19,11 +20,11 @@ public class ManagedReadSessionDemo {
 
   public static void main(String[] args) throws Exception {
 
-    final var authToken = System.getenv("S2_AUTH_TOKEN");
+    final var authToken = System.getenv("S2_ACCESS_TOKEN");
     final var basinName = System.getenv("S2_BASIN");
     final var streamName = System.getenv("S2_STREAM");
     if (authToken == null) {
-      throw new IllegalStateException("S2_AUTH_TOKEN not set");
+      throw new IllegalStateException("S2_ACCESS_TOKEN not set");
     }
     if (basinName == null) {
       throw new IllegalStateException("S2_BASIN not set");
@@ -32,11 +33,7 @@ public static void main(String[] args) throws Exception {
       throw new IllegalStateException("S2_STREAM not set");
     }
 
-    var config =
-        Config.newBuilder(authToken)
-            .withEndpoints(Endpoints.fromEnvironment())
-            .withCompression(true)
-            .build();
+    var config = Config.newBuilder(authToken).withEndpoints(Endpoints.fromEnvironment()).build();
 
     try (final var executor = new ScheduledThreadPoolExecutor(12);
         final var channel = ManagedChannelFactory.forBasinOrStreamService(config, basinName)) {
@@ -49,7 +46,10 @@ public static void main(String[] args) throws Exception {
 
       try (final var managedSession =
           streamClient.managedReadSession(
-              ReadSessionRequest.newBuilder().withHeartbeats(true).build(),
+              ReadSessionRequest.newBuilder()
+                  .withHeartbeats(true)
+                  .withStart(Start.seqNum(0))
+                  .build(),
               1024 * 1024 * 1024 * 5)) {
 
         AtomicLong receivedBytes = new AtomicLong();
@@ -61,10 +61,12 @@ public static void main(String[] args) throws Exception {
                 if (elem instanceof Batch batch) {
                   var size = batch.meteredBytes();
                   logger.info(
-                      "batch of {} bytes, seqnums {}..={}",
+                      "batch of {} bytes, seqnums {}..={} / instants {}..={}",
                       size,
                       batch.firstSeqNum(),
-                      batch.lastSeqNum());
+                      batch.lastSeqNum(),
+                      batch.firstTimestamp(),
+                      batch.lastTimestamp());
                   receivedBytes.addAndGet(size);
                 } else {
                   logger.info("non batch received: {}", elem);

gradle.properties

@@ -1 +1 @@
-version=0.0.15-SNAPSHOT
+version=0.0.15

s2-internal/src/main/proto

@@ -55,6 +55,7 @@ static boolean retryableStatus(Status status) {
     switch (status.getCode()) {
       case UNKNOWN:
       case DEADLINE_EXCEEDED:
+      case RESOURCE_EXHAUSTED:
       case UNAVAILABLE:
         return true;
       default:

s2-sdk/src/main/java/s2/client/BaseClient.java

@@ -20,7 +20,6 @@
 import s2.types.CreateBasinRequest;
 import s2.types.Paginated;
 import s2.types.ReconfigureBasinRequest;
-import s2.types.StreamConfig;
 import s2.v1alpha.AccountServiceGrpc;
 import s2.v1alpha.DeleteBasinRequest;
 import s2.v1alpha.GetBasinConfigRequest;
@@ -125,9 +124,7 @@ public ListenableFuture<BasinConfig> reconfigureBasin(ReconfigureBasinRequest re
                 withStaticRetries(
                     config.maxRetries,
                     () -> this.futureStub.reconfigureBasin(reconfigure.toProto())),
-                resp ->
-                    new BasinConfig(
-                        StreamConfig.fromProto(resp.getConfig().getDefaultStreamConfig())),
+                resp -> BasinConfig.fromProto(resp.getConfig()),
                 executor));
   }
 
@@ -146,9 +143,7 @@ public ListenableFuture<BasinConfig> getBasinConfig(String basin) {
                     () ->
                         this.futureStub.getBasinConfig(
                             GetBasinConfigRequest.newBuilder().setBasin(basin).build())),
-                resp ->
-                    new BasinConfig(
-                        StreamConfig.fromProto(resp.getConfig().getDefaultStreamConfig())),
+                resp -> BasinConfig.fromProto(resp.getConfig()),
                 executor));
   }
 

s2-sdk/src/main/java/s2/client/Client.java

@@ -57,6 +57,10 @@ public class ManagedAppendSession implements AutoCloseable {
     this.remainingAttempts = new AtomicInteger(this.client.config.maxRetries);
   }
 
+  public Integer remainingBufferCapacityBytes() {
+    return this.inflightBytes.availablePermits();
+  }
+
   private ListenableFuture<Void> retryingDaemon() {
     return Futures.catchingAsync(
         executor.submit(this::daemon),
@@ -370,6 +374,8 @@ public ListenableFuture<Void> closeGracefully() {
     return daemon;
   }
 
+  interface Notification {}
+
   static class InflightRecord {
     final AppendInput input;
     final long entryNanos;
@@ -392,8 +398,6 @@ static InflightRecord construct(AppendInput input, Long meteredBytes) {
     }
   }
 
-  interface Notification {}
-
   class Ack implements Notification {
     final AppendOutput output;