[GitHub] [ratis] szetszwo commented on a change in pull request #596: RATIS-1519. When DataStreamManagement#read an exception occurs, remove DataStream

[GitBox <gi...@apache.org>]

szetszwo commented on a change in pull request #596:
URL: https://github.com/apache/ratis/pull/596#discussion_r800408732



##########
File path: ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java
##########
@@ -379,6 +390,7 @@ void read(DataStreamRequestByteBuf request, ChannelHandlerContext ctx,
     try {
       readImpl(request, ctx, buf, getStreams);
     } catch (Throwable t) {
+      removeDataStream(buf);
       buf.release();

Review comment:
       Let's have buf.release() first.
   ```
   @@ -380,6 +392,7 @@ public class DataStreamManagement {
          readImpl(request, ctx, buf, getStreams);
        } catch (Throwable t) {
          buf.release();
   +      removeDataStream(request);
          throw t;
        }
      }
   ```
   

##########
File path: ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java
##########
@@ -240,6 +240,17 @@ StreamInfo remove(ClientInvocationId key) {
     return f;
   }
 
+  private void removeDataStream(ByteBuf buf) {
+    try {
+      RaftClientRequest request = ClientProtoUtils.toRaftClientRequest(
+          RaftClientRequestProto.parseFrom(buf.nioBuffer()));

Review comment:
       This works only if request.getType() == Type.STREAM_HEADER.  The code should be:
   ```
     private void removeDataStream(DataStreamRequestByteBuf request) {
       try {
         final ClientInvocationId invocationId = ClientInvocationId.valueOf(request.getClientId(), request.getStreamId());
         final StreamInfo s = streams.remove(invocationId);
         if (s != null) {
           s.getDivision().getDataStreamMap().remove(invocationId);
         }
       } catch (IOException ignored) {
         LOG.debug(this + ": Failed to removeDataStream for " + request, ignored);
       }
     }
   ```

##########
File path: ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java
##########
@@ -240,6 +240,17 @@ StreamInfo remove(ClientInvocationId key) {
     return f;
   }
 
+  private void removeDataStream(ByteBuf buf) {
+    try {
+      RaftClientRequest request = ClientProtoUtils.toRaftClientRequest(
+          RaftClientRequestProto.parseFrom(buf.nioBuffer()));
+      ClientInvocationId invocationId = ClientInvocationId.valueOf(request);
+      server.getDivision(request.getRaftGroupId()).getDataStreamMap().remove(invocationId);
+    } catch (Throwable e) {
+      throw new CompletionException(e);

Review comment:
       Just print an error message as shown above.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org