updates

sgbalogh · sgbalogh · commit c6c65e9b59ec · 2025-06-02T12:41:06.000-07:00
diff --git a/README.md b/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"
 ```
diff --git a/app/src/main/java/org/example/app/AccountDemo.java b/app/src/main/java/org/example/app/AccountDemo.java
@@ -19,7 +19,7 @@ public class AccountDemo {
 
   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();
 
diff --git a/app/src/main/java/org/example/app/BasinDemo.java b/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();
 
diff --git a/app/src/main/java/org/example/app/ManagedAppendSessionDemo.java b/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()
diff --git a/app/src/main/java/org/example/app/ManagedReadSessionDemo.java b/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);
diff --git a/gradle.properties b/gradle.properties
@@ -1 +1 @@
-version=0.0.15-SNAPSHOT
+version=0.0.15
diff --git a/s2-internal/src/main/proto b/s2-internal/src/main/proto
@@ -1 +1 @@
-Subproject commit aea316a72958fe4ffb00cdb30681d0f5836012c6
+Subproject commit 4183c36ef4b34d8626e01873eafbaf9583aae067
diff --git a/s2-sdk/src/main/java/s2/client/BaseClient.java b/s2-sdk/src/main/java/s2/client/BaseClient.java
@@ -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:
diff --git a/s2-sdk/src/main/java/s2/client/ManagedAppendSession.java b/s2-sdk/src/main/java/s2/client/ManagedAppendSession.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),
diff --git a/s2-sdk/src/main/java/s2/client/ReadSession.java b/s2-sdk/src/main/java/s2/client/ReadSession.java
@@ -10,12 +10,14 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import s2.types.Batch;
 import s2.types.ReadOutput;
 import s2.types.ReadSessionRequest;
+import s2.types.Start;
 import s2.v1alpha.ReadSessionResponse;
 
 public class ReadSession implements AutoCloseable {
@@ -27,7 +29,7 @@ public class ReadSession implements AutoCloseable {
   final ScheduledExecutorService executor;
   final StreamClient client;
 
-  final AtomicLong nextStartSeqNum;
+  final AtomicReference<Start> nextStart;
   final AtomicLong consumedRecords = new AtomicLong();
   final AtomicLong consumedBytes = new AtomicLong(0);
   final AtomicInteger remainingAttempts;
@@ -52,7 +54,7 @@ public class ReadSession implements AutoCloseable {
     this.onResponse = onResponse;
     this.onError = onError;
     this.request = request;
-    this.nextStartSeqNum = new AtomicLong(request.startSeqNum);
+    this.nextStart = new AtomicReference<>(request.start);
     this.remainingAttempts = new AtomicInteger(client.config.maxRetries);
     this.lastEvent = new AtomicLong(System.nanoTime());
 
@@ -134,12 +136,12 @@ private ListenableFuture<Void> retrying() {
 
     return Futures.catchingAsync(
         readSessionInner(
-            request.update(nextStartSeqNum.get(), consumedRecords.get(), consumedBytes.get()),
+            request.update(nextStart.get(), consumedRecords.get(), consumedBytes.get()),
             resp -> {
               if (resp instanceof Batch) {
                 final Batch batch = (Batch) resp;
                 var lastRecordIdx = batch.lastSeqNum();
-                lastRecordIdx.ifPresent(v -> nextStartSeqNum.set(v + 1));
+                lastRecordIdx.ifPresent(v -> nextStart.set(Start.seqNum(v + 1)));
                 consumedRecords.addAndGet(batch.sequencedRecordBatch.records.size());
                 consumedBytes.addAndGet(batch.meteredBytes());
               }
diff --git a/s2-sdk/src/main/java/s2/types/AppendInput.java b/s2-sdk/src/main/java/s2/types/AppendInput.java
@@ -1,17 +1,16 @@
 package s2.types;
 
-import com.google.protobuf.ByteString;
 import java.io.Serializable;
 import java.util.List;
 import java.util.Optional;
 
 public class AppendInput implements MeteredBytes, Serializable {
   public final List<AppendRecord> records;
   public final Optional<Long> matchSeqNum;
-  public final Optional<ByteString> fencingToken;
+  public final Optional<String> fencingToken;
 
   private AppendInput(
-      List<AppendRecord> records, Optional<Long> matchSeqNum, Optional<ByteString> fencingToken) {
+      List<AppendRecord> records, Optional<Long> matchSeqNum, Optional<String> fencingToken) {
     this.records = records;
     this.matchSeqNum = matchSeqNum;
     this.fencingToken = fencingToken;
@@ -38,7 +37,7 @@ public s2.v1alpha.AppendInput toProto(String streamName) {
   public static class AppendInputBuilder {
     private List<AppendRecord> records = List.of();
     private Optional<Long> matchSeqNum = Optional.empty();
-    private Optional<ByteString> fencingToken = Optional.empty();
+    private Optional<String> fencingToken = Optional.empty();
 
     public AppendInputBuilder withRecords(List<AppendRecord> records) {
       this.records = records;
@@ -50,7 +49,7 @@ public AppendInputBuilder withMatchSeqNum(Long matchSeqNum) {
       return this;
     }
 
-    public AppendInputBuilder withFencingToken(ByteString fencingToken) {
+    public AppendInputBuilder withFencingToken(String fencingToken) {
       this.fencingToken = Optional.of(fencingToken);
       return this;
     }
@@ -64,8 +63,8 @@ public AppendInput build() {
           });
       fencingToken.ifPresent(
           token -> {
-            if (token.size() > 16) {
-              throw new IllegalArgumentException("fencingToken must be less than 16 bytes");
+            if (token.length() > 36) {
+              throw new IllegalArgumentException("fencingToken must be 36 or fewer UTF-8 bytes");
             }
           });
       var provisional = new AppendInput(records, matchSeqNum, fencingToken);
diff --git a/s2-sdk/src/main/java/s2/types/Batch.java b/s2-sdk/src/main/java/s2/types/Batch.java
@@ -1,5 +1,6 @@
 package s2.types;
 
+import java.time.Instant;
 import java.util.Optional;
 
 public final class Batch implements ReadOutput, MeteredBytes {
@@ -23,6 +24,19 @@ public Optional<Long> lastSeqNum() {
     }
   }
 
+  public Optional<Instant> firstTimestamp() {
+    return this.sequencedRecordBatch.records.stream().findFirst().map(sr -> sr.timestamp);
+  }
+
+  public Optional<Instant> lastTimestamp() {
+    if (!this.sequencedRecordBatch.records.isEmpty()) {
+      return Optional.of(
+          this.sequencedRecordBatch.records.get(sequencedRecordBatch.records.size() - 1).timestamp);
+    } else {
+      return Optional.empty();
+    }
+  }
+
   @Override
   public long meteredBytes() {
     return this.sequencedRecordBatch.meteredBytes();
diff --git a/s2-sdk/src/main/java/s2/types/ReadOutput.java b/s2-sdk/src/main/java/s2/types/ReadOutput.java
@@ -5,11 +5,9 @@ static ReadOutput fromProto(s2.v1alpha.ReadOutput readOutput) {
     switch (readOutput.getOutputCase()) {
       case BATCH:
         return new Batch(SequencedRecordBatch.fromProto(readOutput.getBatch()));
-      case FIRST_SEQ_NUM:
-        return new FirstSeqNum(readOutput.getFirstSeqNum());
       case NEXT_SEQ_NUM:
         return new NextSeqNum(readOutput.getNextSeqNum());
     }
     throw new IllegalStateException("Unrecognized readOutput case: " + readOutput);
   }
-}
+}
diff --git a/s2-sdk/src/main/java/s2/types/ReadRequest.java b/s2-sdk/src/main/java/s2/types/ReadRequest.java
@@ -4,11 +4,11 @@
 
 public class ReadRequest {
 
-  public final long startSeqNum;
+  public final Start start;
   public final ReadLimit readLimit;
 
-  protected ReadRequest(long startSeqNum, ReadLimit readLimit) {
-    this.startSeqNum = startSeqNum;
+  protected ReadRequest(Start start, ReadLimit readLimit) {
+    this.start = start;
     this.readLimit = readLimit;
   }
 
@@ -17,19 +17,28 @@ public static ReadRequestBuilder newBuilder() {
   }
 
   public s2.v1alpha.ReadRequest toProto(String streamName) {
-    return s2.v1alpha.ReadRequest.newBuilder()
-        .setStream(streamName)
-        .setStartSeqNum(startSeqNum)
-        .setLimit(readLimit.toProto())
-        .build();
+    s2.v1alpha.ReadRequest.Builder builder =
+        s2.v1alpha.ReadRequest.newBuilder().setStream(streamName).setLimit(readLimit.toProto());
+
+    if (start instanceof Start.SeqNum) {
+      builder.setSeqNum(((Start.SeqNum) start).value);
+    } else if (start instanceof Start.Timestamp) {
+      builder.setTimestamp(((Start.Timestamp) start).value);
+    } else if (start instanceof Start.TailOffset) {
+      builder.setTailOffset(((Start.TailOffset) start).value);
+    } else {
+      throw new IllegalStateException("Unknown start type: " + start.getClass().getName());
+    }
+
+    return builder.build();
   }
 
   public static class ReadRequestBuilder {
-    private Optional<Long> startSeqNum = Optional.empty();
+    private Optional<Start> start = Optional.empty();
     private Optional<ReadLimit> readLimit = Optional.empty();
 
-    public ReadRequestBuilder withStartSeqNum(long startSeqNum) {
-      this.startSeqNum = Optional.of(startSeqNum);
+    public ReadRequestBuilder withStart(Start start) {
+      this.start = Optional.of(start);
       return this;
     }
 
@@ -40,7 +49,8 @@ public ReadRequestBuilder withReadLimit(ReadLimit readLimit) {
 
     public ReadRequest build() {
       this.readLimit.ifPresent(ReadLimit::validateUnary);
-      return new ReadRequest(this.startSeqNum.orElse(0L), this.readLimit.orElse(ReadLimit.NONE));
+      return new ReadRequest(
+          this.start.orElseGet(() -> Start.seqNum(0L)), this.readLimit.orElse(ReadLimit.NONE));
     }
   }
 }
diff --git a/s2-sdk/src/main/java/s2/types/ReadSessionRequest.java b/s2-sdk/src/main/java/s2/types/ReadSessionRequest.java
diff --git a/s2-sdk/src/main/java/s2/types/SequencedRecord.java b/s2-sdk/src/main/java/s2/types/SequencedRecord.java
diff --git a/s2-sdk/src/main/java/s2/types/Start.java b/s2-sdk/src/main/java/s2/types/Start.java
diff --git a/s2-sdk/src/test/java/s2/client/ReadSessionTest.java b/s2-sdk/src/test/java/s2/client/ReadSessionTest.java
diff --git a/s2-sdk/src/test/java/s2/v1alpha/StreamService/MockReadSessionStreamService.java b/s2-sdk/src/test/java/s2/v1alpha/StreamService/MockReadSessionStreamService.java

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-version=0.0.15-SNAPSHOT`
	`1`	`+version=0.0.15`