HDDS-15013. gRPC channel holding objects on completed request in ReadBlock (#10071)

Author: ss77892

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java

@@ -54,6 +54,7 @@
 import org.apache.ratis.protocol.exceptions.TimeoutIOException;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
+import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver;
 import org.apache.ratis.util.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -65,6 +66,7 @@
 public class StreamBlockInputStream extends BlockExtendedInputStream {
   private static final Logger LOG = LoggerFactory.getLogger(StreamBlockInputStream.class);
   private static final int EOF = -1;
+  private static final String STREAM_CLOSE_REASON = "StreamBlockInputStream closed";
   private static final AtomicInteger STREAM_ID = new AtomicInteger(0);
   private static final AtomicInteger READER_ID = new AtomicInteger(0);
 
@@ -225,12 +227,36 @@ public synchronized void unbuffer() {
   }
 
   private synchronized void closeStream() {
-    if (streamingReader != null) {
-      LOG.debug("Closing {}", streamingReader);
-      streamingReader.onCompleted();
-      streamingReader = null;
+    if (streamingReader == null) {
+      buffer = null;
+      return;
     }
+
+    final StreamingReader reader = streamingReader;
+    streamingReader = null;
     buffer = null;
+
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Closing {}", reader);
+    }
+
+    reader.onCompleted();
+
+    final StreamingReadResponse response = reader.getResponse();
+    if (response != null) {
+      final ClientCallStreamObserver<ContainerProtos.ContainerCommandRequestProto> requestObserver =
+          response.getRequestObserver();
+      try {
+        requestObserver.onCompleted();
+      } catch (RuntimeException e) {
+        LOG.warn("Failed to close gRPC request stream for {}", reader, e);
+        try {
+          requestObserver.cancel(STREAM_CLOSE_REASON, e);
+        } catch (RuntimeException cancelEx) {
+          LOG.warn("Failed to cancel gRPC request stream for {}", reader, cancelEx);
+        }
+      }
+    }
   }
 
   protected synchronized void checkOpen() throws IOException {

hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java

@@ -19,8 +19,10 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -50,6 +52,7 @@
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
 import org.apache.hadoop.security.token.Token;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver;
 import org.junit.jupiter.api.Test;
 
 /**
@@ -97,7 +100,8 @@ public void testReleasesStreamPermitAtBlockEof() throws Exception {
     byte[] data = new byte[] {1, 2, 3, 4};
     long length = data.length;
     Pipeline pipeline = mockStandalonePipeline();
-    XceiverClientGrpc xceiverClient = mockStreamingReadClient(data);
+    ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver = mock(ClientCallStreamObserver.class);
+    XceiverClientGrpc xceiverClient = mockStreamingReadClient(data, requestObserver);
     XceiverClientFactory xceiverClientFactory = mock(XceiverClientFactory.class);
     when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class)))
         .thenReturn(xceiverClient);
@@ -115,13 +119,73 @@ public void testReleasesStreamPermitAtBlockEof() throws Exception {
       assertEquals(data[(int) length - 1] & 0xFF, last);
       assertEquals(length, sbis.getPos());
       verify(xceiverClient, times(1)).completeStreamRead();
+      verify(requestObserver, times(1)).onCompleted();
 
       // Subsequent reads should return EOF and must not trigger duplicate permit release.
       assertEquals(-1, sbis.read());
       assertEquals(-1, sbis.read());
     }
 
     verify(xceiverClient, times(1)).completeStreamRead();
+    verify(requestObserver, times(1)).onCompleted();
+    verify(requestObserver, never()).cancel(any(), any());
+  }
+
+  @Test
+  public void testCancelsRequestStreamWhenOnCompletedThrows() throws Exception {
+    OzoneClientConfig clientConfig = newStreamReadConfig();
+    BlockID blockID = new BlockID(1L, 3L);
+    byte[] data = new byte[] {1, 2, 3, 4};
+    Pipeline pipeline = mockStandalonePipeline();
+    ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver = mock(ClientCallStreamObserver.class);
+    RuntimeException closeFailure = new RuntimeException("close failed");
+    doThrow(closeFailure).when(requestObserver).onCompleted();
+
+    XceiverClientGrpc xceiverClient = mockStreamingReadClient(data, requestObserver);
+    XceiverClientFactory xceiverClientFactory = mock(XceiverClientFactory.class);
+    when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class))).thenReturn(xceiverClient);
+
+    try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+        blockID, data.length, pipeline, null, xceiverClientFactory, NO_REFRESH, clientConfig)) {
+      ByteBuffer all = ByteBuffer.allocate(data.length);
+      assertEquals(data.length, sbis.read(all));
+      assertEquals(data.length, sbis.getPos());
+      assertEquals(-1, sbis.read());
+    }
+
+    verify(requestObserver, times(1)).onCompleted();
+    verify(requestObserver, times(1)).cancel(eq("StreamBlockInputStream closed"), eq(closeFailure));
+    verify(xceiverClient, times(1)).completeStreamRead();
+  }
+
+  @Test
+  public void testCloseDoesNotFailWhenOnCompletedAndCancelThrow() throws Exception {
+    OzoneClientConfig clientConfig = newStreamReadConfig();
+    BlockID blockID = new BlockID(1L, 4L);
+    byte[] data = new byte[] {1, 2, 3, 4};
+    Pipeline pipeline = mockStandalonePipeline();
+    ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver = mock(ClientCallStreamObserver.class);
+    RuntimeException closeFailure = new RuntimeException("close failed");
+    RuntimeException cancelFailure = new RuntimeException("cancel failed");
+    doThrow(closeFailure).when(requestObserver).onCompleted();
+    doThrow(cancelFailure).when(requestObserver)
+        .cancel(eq("StreamBlockInputStream closed"), eq(closeFailure));
+
+    XceiverClientGrpc xceiverClient = mockStreamingReadClient(data, requestObserver);
+    XceiverClientFactory xceiverClientFactory = mock(XceiverClientFactory.class);
+    when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class))).thenReturn(xceiverClient);
+
+    try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+        blockID, data.length, pipeline, null, xceiverClientFactory, NO_REFRESH, clientConfig)) {
+      ByteBuffer all = ByteBuffer.allocate(data.length);
+      assertEquals(data.length, sbis.read(all));
+      assertEquals(data.length, sbis.getPos());
+      assertEquals(-1, sbis.read());
+    }
+
+    verify(requestObserver, times(1)).onCompleted();
+    verify(requestObserver, times(1)).cancel(eq("StreamBlockInputStream closed"), eq(closeFailure));
+    verify(xceiverClient, times(1)).completeStreamRead();
   }
 
   private OzoneClientConfig newStreamReadConfig() {
@@ -149,10 +213,12 @@ private Pipeline mockStandalonePipeline() throws Exception {
     return pipeline;
   }
 
-  private XceiverClientGrpc mockStreamingReadClient(byte[] data) throws Exception {
+  private XceiverClientGrpc mockStreamingReadClient(byte[] data,
+      ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver) throws Exception {
     XceiverClientGrpc xceiverClient = mock(XceiverClientGrpc.class);
     StreamingReadResponse streamingReadResponse = mock(StreamingReadResponse.class);
     ReadBlockResponseProto readBlock = buildReadBlockResponse(data);
+    when(streamingReadResponse.getRequestObserver()).thenReturn(requestObserver);
 
     doNothing().when(xceiverClient)
         .streamRead(any(ContainerCommandRequestProto.class),