chore: lang version to 11 (#12)

Author: sgbalogh

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

@@ -26,7 +26,7 @@ public static void main(String[] args) throws Exception {
     try (var client = Client.newBuilder(config).build()) {
 
       var basins = client.listBasins(ListBasinsRequest.newBuilder().build()).get();
-      basins.elems().forEach(basin -> logger.info("basin={}", basin));
+      basins.elems.forEach(basin -> logger.info("basin={}", basin));
 
       var newBasin =
           client

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

@@ -22,12 +22,10 @@ public static void main(String[] args) throws Exception {
     try (final var basinClient =
         BasinClient.newBuilder(config, System.getenv("S2_BASIN")).build()) {
       final var streams = basinClient.listStreams(ListStreamsRequest.newBuilder().build()).get();
-      streams
-          .elems()
-          .forEach(
-              stream -> {
-                logger.info("stream={}", stream);
-              });
+      streams.elems.forEach(
+          stream -> {
+            logger.info("stream={}", stream);
+          });
 
       var newStream =
           basinClient

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

@@ -33,9 +33,13 @@ public static void main(String[] args) throws Exception {
       throw new IllegalStateException("S2_STREAM not set");
     }
 
-    var config = Config.newBuilder(authToken).withEndpoints(Endpoints.fromEnvironment()).build();
+    var config =
+        Config.newBuilder(authToken)
+            .withEndpoints(Endpoints.fromEnvironment())
+            .withCompression(true)
+            .build();
 
-    try (final var executor = new ScheduledThreadPoolExecutor(1);
+    try (final var executor = new ScheduledThreadPoolExecutor(12);
         final var channel = ManagedChannelFactory.forBasinOrStreamService(config, basinName)) {
 
       final var streamClient =
@@ -47,7 +51,7 @@ public static void main(String[] args) throws Exception {
       try (final var managedSession =
           streamClient.managedReadSession(
               ReadSessionRequest.newBuilder().withReadLimit(ReadLimit.count(100_000)).build(),
-              1024 * 1024 * 1024)) {
+              1024 * 1024 * 1024 * 5)) {
 
         AtomicLong receivedBytes = new AtomicLong();
         while (!managedSession.isClosed()) {

gradle.properties

@@ -1 +1 @@
-version=0.0.13-SNAPSHOT
+version=0.0.13

s2-internal/build.gradle.kts

@@ -54,7 +54,7 @@ protobuf {
 
 java {
     toolchain {
-        languageVersion.set(JavaLanguageVersion.of(17))
+        languageVersion.set(JavaLanguageVersion.of(11))
     }
     withJavadocJar()
     withSourcesJar()

s2-sdk/build.gradle.kts

@@ -45,7 +45,7 @@ tasks.test {
 
 java {
     toolchain {
-        languageVersion.set(JavaLanguageVersion.of(17))
+        languageVersion.set(JavaLanguageVersion.of(11))
     }
     withJavadocJar()
     withSourcesJar()

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

@@ -52,10 +52,14 @@ public Thread newThread(Runnable r) {
   }
 
   static boolean retryableStatus(Status status) {
-    return switch (status.getCode()) {
-      case UNKNOWN, DEADLINE_EXCEEDED, UNAVAILABLE -> true;
-      default -> false;
-    };
+    switch (status.getCode()) {
+      case UNKNOWN:
+      case DEADLINE_EXCEEDED:
+      case UNAVAILABLE:
+        return true;
+      default:
+        return false;
+    }
   }
 
   public void close() {

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

@@ -8,6 +8,7 @@
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
 import s2.auth.BearerTokenCallCredentials;
 import s2.channel.BasinCompatibleChannel;
 import s2.channel.ManagedChannelFactory;
@@ -66,7 +67,9 @@ public ListenableFuture<Paginated<StreamInfo>> listStreams(
                 resp ->
                     new Paginated<>(
                         resp.getHasMore(),
-                        resp.getStreamsList().stream().map(StreamInfo::fromProto).toList()),
+                        resp.getStreamsList().stream()
+                            .map(StreamInfo::fromProto)
+                            .collect(Collectors.toList())),
                 executor));
   }
 

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

@@ -8,6 +8,7 @@
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import s2.auth.BearerTokenCallCredentials;
@@ -61,7 +62,9 @@ public ListenableFuture<Paginated<BasinInfo>> listBasins(s2.types.ListBasinsRequ
                 resp ->
                     new Paginated<>(
                         resp.getHasMore(),
-                        resp.getBasinsList().stream().map(BasinInfo::fromProto).toList()),
+                        resp.getBasinsList().stream()
+                            .map(BasinInfo::fromProto)
+                            .collect(Collectors.toList())),
                 executor));
   }
 

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

@@ -179,17 +179,19 @@ record -> {
         throw Status.CANCELLED
             .withDescription("hit deadline while retransmitting")
             .asRuntimeException();
-      } else if (notification instanceof Ack ack) {
+      } else if (notification instanceof Ack) {
+        final Ack ack = (Ack) notification;
         this.remainingAttempts.set(this.client.config.maxRetries);
         var correspondingInflight = inflightQueue.poll();
         if (correspondingInflight == null) {
           throw Status.INTERNAL.withDescription("inflight queue is empty").asRuntimeException();
         } else {
           validate(correspondingInflight, ack.output);
           correspondingInflight.callback.set(ack.output);
-          this.inflightBytes.release(correspondingInflight.meteredBytes.intValue());
+          this.inflightBytes.release((int) correspondingInflight.meteredBytes);
         }
-      } else if (notification instanceof Error error) {
+      } else if (notification instanceof Error) {
+        final Error error = (Error) notification;
         throw new RuntimeException(error.throwable);
       } else {
         throw Status.INTERNAL
@@ -218,7 +220,8 @@ private synchronized Void cleanUp(Status fatal) throws InterruptedException {
     // through them and cancel any pending batches via their callback.
     while (!notificationQueue.isEmpty()) {
       var entry = notificationQueue.poll();
-      if (entry instanceof Batch batch) {
+      if (entry instanceof Batch) {
+        final Batch batch = (Batch) entry;
         batch.input.callback.setException(fatal.asRuntimeException());
       }
     }
@@ -228,7 +231,7 @@ private synchronized Void cleanUp(Status fatal) throws InterruptedException {
   }
 
   private void validate(InflightRecord record, AppendOutput output) {
-    var numRecordsForAcknowledgement = output.endSeqNum() - output.startSeqNum();
+    var numRecordsForAcknowledgement = output.endSeqNum - output.startSeqNum;
     if (numRecordsForAcknowledgement != record.input.records.size()) {
       throw Status.INTERNAL
           .withDescription(
@@ -287,7 +290,8 @@ public void onCompleted() {
           clientObserver.onError(Status.CANCELLED.asRuntimeException());
           throw Status.CANCELLED.asRuntimeException();
         }
-      } else if (notification instanceof Batch batch) {
+      } else if (notification instanceof Batch) {
+        final Batch batch = (Batch) notification;
         logger.debug("notification=BATCH");
         if (!inflightQueue.offer(batch.input)) {
           throw Status.INTERNAL.asRuntimeException();
@@ -304,7 +308,8 @@ public void onCompleted() {
                       + TimeUnit.NANOSECONDS.convert(this.client.config.requestTimeout)));
         }
 
-      } else if (notification instanceof Ack ack) {
+      } else if (notification instanceof Ack) {
+        final Ack ack = (Ack) notification;
         logger.debug("notification=ACK");
         this.remainingAttempts.set(this.client.config.maxRetries);
         var correspondingInflight = inflightQueue.poll();
@@ -313,7 +318,7 @@ public void onCompleted() {
         } else {
           validate(correspondingInflight, ack.output);
           correspondingInflight.callback.set(ack.output);
-          this.inflightBytes.release(correspondingInflight.meteredBytes.intValue());
+          this.inflightBytes.release((int) correspondingInflight.meteredBytes);
 
           // Reset the next deadline.
           this.nextDeadlineSystemNanos.set(
@@ -323,18 +328,20 @@ public void onCompleted() {
                           entry.entryNanos
                               + TimeUnit.NANOSECONDS.convert(this.client.config.requestTimeout)));
         }
-      } else if (notification instanceof Error error) {
+      } else if (notification instanceof Error) {
+        final Error error = (Error) notification;
         logger.debug("notification=ERROR");
         clientObserver.onError(Status.CANCELLED.asRuntimeException());
         throw new RuntimeException(error.throwable);
 
-      } else if (notification instanceof ClientClose close) {
+      } else if (notification instanceof ClientClose) {
+        final ClientClose close = (ClientClose) notification;
         logger.debug("notification=CLIENT_CLOSE,gracefully={}", close.gracefully);
         clientObserver.onCompleted();
         if (!close.gracefully) {
           return null;
         }
-      } else if (notification instanceof ServerClose close) {
+      } else if (notification instanceof ServerClose) {
         logger.debug("notification=SERVER_CLOSE");
         if (acceptingAppends.get() || !inflightQueue.isEmpty() || !notificationQueue.isEmpty()) {
           throw Status.INTERNAL
@@ -363,25 +370,63 @@ public ListenableFuture<Void> closeGracefully() {
     return daemon;
   }
 
-  sealed interface Notification permits Ack, Batch, ClientClose, Error, ServerClose {}
+  static class InflightRecord {
+    final AppendInput input;
+    final long entryNanos;
+    final SettableFuture<AppendOutput> callback;
+    final long meteredBytes;
+
+    InflightRecord(
+        AppendInput input,
+        long entryNanos,
+        SettableFuture<AppendOutput> callback,
+        long meteredBytes) {
+      this.input = input;
+      this.entryNanos = entryNanos;
+      this.callback = callback;
+      this.meteredBytes = meteredBytes;
+    }
 
-  record InflightRecord(
-      AppendInput input,
-      Long entryNanos,
-      SettableFuture<AppendOutput> callback,
-      Long meteredBytes) {
     static InflightRecord construct(AppendInput input, Long meteredBytes) {
       return new InflightRecord(input, System.nanoTime(), SettableFuture.create(), meteredBytes);
     }
   }
 
-  record Batch(InflightRecord input) implements Notification {}
+  interface Notification {}
+
+  class Ack implements Notification {
+    final AppendOutput output;
+
+    Ack(AppendOutput output) {
+      this.output = output;
+    }
+  }
 
-  record Ack(AppendOutput output) implements Notification {}
+  class Batch implements Notification {
+    final InflightRecord input;
 
-  record Error(Throwable throwable) implements Notification {}
+    Batch(InflightRecord input) {
+      this.input = input;
+    }
+  }
 
-  record ClientClose(boolean gracefully) implements Notification {}
+  class ClientClose implements Notification {
+    final boolean gracefully;
 
-  record ServerClose() implements Notification {}
+    ClientClose(boolean gracefully) {
+      this.gracefully = gracefully;
+    }
+  }
+
+  class Error implements Notification {
+    final Throwable throwable;
+
+    Error(Throwable throwable) {
+      this.throwable = throwable;
+    }
+  }
+
+  class ServerClose implements Notification {
+    ServerClose() {}
+  }
 }