feat(stream): fail pending request if server close stream without

exception

client/src/main/java/io/streamnative/oxia/client/batch/ReadBatch.java

@@ -23,6 +23,8 @@
 import io.streamnative.oxia.proto.ReadResponse;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CancellationException;
+
 import lombok.NonNull;
 
 final class ReadBatch extends BatchBase implements Batch, StreamObserver<ReadResponse> {
@@ -83,6 +85,12 @@ public void onError(Throwable batchError) {
 
     @Override
     public void onCompleted() {
+        // complete pending request if the server close stream without any response
+        gets.forEach(g -> {
+            if (!g.callback().isDone()) {
+                g.fail(new CancellationException());
+            }
+        });
         factory.getReadRequestLatencyHistogram().recordSuccess(System.nanoTime() - startSendTimeNanos);
     }
 

client/src/main/java/io/streamnative/oxia/client/grpc/WriteStreamWrapper.java

@@ -22,6 +22,7 @@
 import io.streamnative.oxia.proto.WriteResponse;
 import java.util.ArrayDeque;
 import java.util.Deque;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import lombok.extern.slf4j.Slf4j;
 
@@ -63,7 +64,16 @@ public void onError(Throwable t) {
     }
 
     @Override
-    public void onCompleted() {}
+    public void onCompleted() {
+        synchronized (WriteStreamWrapper.this) {
+            // complete pending request if the server close stream without any response
+            pendingWrites.forEach(f -> {
+                if (!f.isDone()) {
+                    f.completeExceptionally(new CancellationException());
+                }
+            });
+        }
+    }
 
     public CompletableFuture<WriteResponse> send(WriteRequest request) {
         synchronized (WriteStreamWrapper.this) {