Added a soft reference based shared cache for S3 reads

extensions/s3/src/main/java/io/deephaven/extensions/s3/S3ChannelContext.java

@@ -304,7 +304,7 @@ public boolean equals(Object obj) {
         private final long from;
         private final long to;
         private final Instant createdAt;
-        private volatile CompletableFuture<Boolean> consumerFuture;
+        private CompletableFuture<Boolean> consumerFuture;
         private volatile CompletableFuture<Boolean> producerFuture;
         private int fillCount;
         private long fillBytes;
@@ -433,7 +433,7 @@ public void cleanup() {
         public void accept(final Boolean isComplete, final Throwable throwable) {
             if (log.isDebugEnabled()) {
                 final Instant completedAt = Instant.now();
-                if (isComplete) {
+                if (Boolean.TRUE.equals(isComplete)) {
                     log.debug().append("Send complete: ").append(requestStr()).append(' ')
                             .append(Duration.between(createdAt, completedAt).toString()).endl();
                 } else {
@@ -474,12 +474,12 @@ private ByteBuffer getFullFragment() throws ExecutionException, InterruptedExcep
             // apiCallTimeout.
             final long readNanos = instructions.readTimeout().plusMillis(100).toNanos();
             final Boolean isComplete = consumerFuture.get(readNanos, TimeUnit.NANOSECONDS);
-            if (!isComplete) {
+            if (Boolean.FALSE.equals(isComplete)) {
                 throw new UncheckedDeephavenException(String.format("Failed to complete request %s", requestStr()));
             }
             final ByteBuffer result = get();
             if (result == null) {
-                throw new IllegalStateException(
+                throw new UncheckedDeephavenException(
                         String.format("Failed to acquire buffer after completion, %s", requestStr()));
             }
             if (result.position() != 0 || result.limit() != result.capacity() || result.limit() != requestLength()) {
@@ -532,6 +532,7 @@ private final class Sub implements Subscriber<ByteBuffer> {
                     return;
                 }
                 if (buffer.position() != 0) {
+                    // We don't change the buffer position while writing to it, so this should never happen
                     localProducer.completeExceptionally(new IllegalStateException(
                             String.format("Buffer not empty for new subscriber, %s", requestStr())));
                 }
@@ -554,15 +555,15 @@ public void onSubscribe(Subscription s) {
             }
 
             @Override
-            public void onNext(final ByteBuffer byteBuffer) {
-                final ByteBuffer buffer = Request.this.get();
-                if (buffer == null) {
+            public void onNext(final ByteBuffer dataBuffer) {
+                final ByteBuffer resultBuffer = Request.this.get();
+                if (resultBuffer == null) {
                     localProducer.completeExceptionally(new IllegalStateException(
                             String.format("Failed to acquire buffer for data, %s", requestStr())));
                     return;
                 }
-                final int numBytes = byteBuffer.remaining();
-                buffer.duplicate().position(offset).put(byteBuffer);
+                final int numBytes = dataBuffer.remaining();
+                resultBuffer.duplicate().position(offset).put(dataBuffer);
                 offset += numBytes;
                 subscription.request(1);
             }

extensions/s3/src/main/java/io/deephaven/extensions/s3/S3RequestCache.java

@@ -57,36 +57,37 @@ int getFragmentSize() {
      * @return the request
      */
     @NotNull
-    S3ChannelContext.Request.AcquiredRequest getOrCreateRequest(@NotNull final S3Uri uri, final long fragmentIndex,
+    Request.AcquiredRequest getOrCreateRequest(@NotNull final S3Uri uri, final long fragmentIndex,
             @NotNull final S3ChannelContext context) {
         final Request.ID key = new Request.ID(uri, fragmentIndex);
         Request.AcquiredRequest newAcquiredRequest = null;
-        Request existingRequest = requests.get(key);
-        for (int retryCount = 0; retryCount < Integer.MAX_VALUE; retryCount++) {
+        while (true) {
+            final Request existingRequest = requests.get(key);
             if (existingRequest != null) {
                 final Request.AcquiredRequest acquired = existingRequest.tryAcquire();
                 if (acquired != null) {
                     return acquired;
-                } else {
-                    remove(existingRequest);
                 }
             }
             if (newAcquiredRequest == null) {
                 newAcquiredRequest = Request.createAndAcquire(fragmentIndex, context);
             }
-            existingRequest = requests.putIfAbsent(key, newAcquiredRequest.request);
+            final boolean added;
             if (existingRequest == null) {
+                // Ideally, we could have used ".replace" in this case as well, but KeyedObjectHashMap.replace currently
+                // has a bug when the key is not present in the map.
+                added = requests.putIfAbsent(key, newAcquiredRequest.request) == null;
+            } else {
+                added = requests.replace(key, existingRequest, newAcquiredRequest.request);
+            }
+            if (added) {
                 if (log.isDebugEnabled()) {
                     log.debug().append("Added new request to cache: ").append(String.format("ctx=%d ",
                             System.identityHashCode(context))).append(newAcquiredRequest.request.requestStr()).endl();
                 }
                 return newAcquiredRequest;
             }
         }
-        // We have tried to add the request to the cache too many times
-        throw new IllegalStateException(
-                String.format("Failed to add request to cache: ctx=%d, uri=%s, fragmentIndex=%d",
-                        System.identityHashCode(context), uri, fragmentIndex));
     }
 
     /**