diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java index 79bfc4d9ad..18ffd7f193 100644 --- a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java +++ b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java @@ -40,6 +40,8 @@ public class SubStreamHandler extends Thread implements Serializable { private final transient CountDownLatch latch = new CountDownLatch(1); + private volatile boolean isClosed = false; + private final transient ConsumerServiceStub consumerAsyncClient; private final transient EventMeshGrpcClientConfig clientConfig; @@ -130,6 +132,12 @@ public void run() { } public void close() { + // Avoid closing multiple times + if (isClosed) { + return; + } + isClosed = true; + if (this.sender != null) { senderOnComplete(); }