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 d34a0c79b1..79bfc4d9ad 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
@@ -96,6 +96,7 @@ public void onNext(final CloudEvent message) {
             @Override
             public void onError(final Throwable t) {
                 log.error("Received Server side error", t);
+                close();
             }
 
             @Override
@@ -145,6 +146,7 @@ private void senderOnNext(final CloudEvent subscription) {
             }
         } catch (Exception e) {
             log.error("StreamObserver Error onNext", e);
+            close();
         }
     }
 
@@ -152,6 +154,7 @@ private void senderOnComplete() {
         try {
             synchronized (sender) {
                 sender.onCompleted();
+                sender = null;
             }
         } catch (Exception e) {
             log.error("StreamObserver Error onComplete", e);