From d84d098f3e0f9a234ba913f516e0807d11bcdfbd Mon Sep 17 00:00:00 2001
From: wucheng1997 <jhs1997@qq.com>
Date: Mon, 10 Oct 2022 19:13:53 +0800
Subject: [PATCH] fix dubbo proxyless xds connect, fix Connection closed after
 GOAWAY, add xds observer retry create (#10544)

---
 .../xds/util/protocol/AbstractProtocol.java         | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
index 9b6687f54b..eb21cf3548 100644
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
+++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
@@ -154,6 +154,11 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
                 // observer reused
                 StreamObserver<DiscoveryRequest> observer = requestObserverMap.get(request);
 
+                if (observer == null) {
+                    observer = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
+                    requestObserverMap.put(request, observer);
+                }
+
                 // send request to control panel
                 observer.onNext(buildDiscoveryRequest(names));
 
@@ -231,11 +236,17 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
         @Override
         public void onError(Throwable t) {
             logger.error("xDS Client received error message! detail:", t);
+            clear();
         }
 
         @Override
         public void onCompleted() {
-            // ignore
+            logger.info("xDS Client completed, requestId: " + requestId);
+            clear();
+        }
+
+        private void clear() {
+            requestObserverMap.remove(requestId);
         }
     }