del watcher from watchers when canceled

Signed-off-by: chaofengliu-okg <[email protected]>

Author: chaofengliu-okg

jetcd-core/src/main/java/io/etcd/jetcd/impl/WatchImpl.java

@@ -226,13 +226,20 @@ public void close() {
 
         @Override
         public void cancel() {
-            WriteStream<WatchRequest> ws = wstream.get();
-            if (ws instanceof GrpcWriteStream<?>) {
-                GrpcWriteStream<?> gws = (GrpcWriteStream<?>) ws;
-                var observer = gws.streamObserver();
-                if (observer instanceof ClientCallStreamObserver<?>) {
-                    ClientCallStreamObserver<?> callObs = (ClientCallStreamObserver<?>) observer;
-                    callObs.cancel("Watcher cancelled", null);
+            synchronized (WatchImpl.this.lock) {
+                if (closed.compareAndSet(false, true)) {
+                    WriteStream<WatchRequest> ws = wstream.get();
+                    if (ws instanceof GrpcWriteStream<?>) {
+                        GrpcWriteStream<?> gws = (GrpcWriteStream<?>) ws;
+                        var observer = gws.streamObserver();
+                        if (observer instanceof ClientCallStreamObserver<?>) {
+                            ClientCallStreamObserver<?> callObs = (ClientCallStreamObserver<?>) observer;
+                            callObs.cancel("Watcher cancelled", null);
+                        }
+                    }
+                    id = -1;
+                    listener.onCompleted();
+                    watchers.remove(this);
                 }
             }
         }