fix dubbo proxyless xds connect, fix Connection closed after GOAWAY, …

…add xds observer retry create (apache#10544)

dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java

@@ -154,6 +154,11 @@ public long observeResource(Set<String> resourceNames, Consumer<T> consumer) {
                 // 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 void onNext(DiscoveryResponse value) {
         @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);
         }
     }