From b8b9be010347b760b4a04c34f06f5516d1881504 Mon Sep 17 00:00:00 2001 From: Bram Date: Thu, 8 Oct 2026 08:00:02 +0200 Subject: [PATCH] xDS: fix DistributorWatcher thread safety in CertificateProvider (#13087) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A gRPC client using xDS-driven mTLS does two things concurrently: 1. Certificate provider: watches cert files on disk, periodically reloads them, and notifies subscribers when certs change. 2. Channel subscriptions: each gRPC channel that needs mTLS registers a watcher on the cert provider to get the TLS credentials. The `DistributorWatcher` sits in the middle and fans out: one cert provider pushes updates to it, and it distributes to N channel watchers. I think there are two conditions in this that can lead to a `ConcurrentModificationException`: 1. `updateSpiffeTrustMap` isn't synchronized (unlike `updateCertificate` and `updateTrustedRoots`), so it can race against `addWatcher`/`removeWatcher` from another thread. 2. if callbacks mutate their own subscription. A callchain (line numbers per 1.83.0 / 1084e9707) for that is: ``` DistributorWatcher.onError() CertificateProvider.java:126 → for (watcher : downstreamWatchers) CertificateProvider.java:127 → watcher.onError(errorStatus) CertificateProvider.java:128 → DynamicSslContextProvider.onError() DynamicSslContextProvider.java:143 → callback.onException(error) DynamicSslContextProvider.java:145 → SslContextProviderSupplier.onException() SslContextProviderSupplier.java:79 → releaseSslContextProvider(toRelease) SslContextProviderSupplier.java:81 → TlsContextManagerImpl.releaseClient...() TlsContextManagerImpl.java:78 → ReferenceCountingMap.releaseInternal() ReferenceCountingMap.java:89 → if refcount == 0: value.close() ReferenceCountingMap.java:95 → CertProviderSslContextProvider.close() CertProviderSslContextProvider.java:216 → certHandle.close() CertProviderSslContextProvider.java:218 → Handle.close() CertificateProviderStore.java:62 → distWatcher.removeWatcher() CertificateProviderStore.java:64 <-- REENTRANT ``` We observed `ConcurrentModificationException` in `updateSpiffeTrustMap()` in production, consistent with the unsynchronized iteration described above. We also observed startup RPC deadlines during the rollout, but have not established the causal connection between these symptoms. These all occurred in a service that creates many (~60) channels on startup, while getting the CDS config over xDS. The service uses grpc-java / grpc-xds `1.83.0`. When we switched it to use mTLS with SPIFFE certificates, we got the following errors: ``` DEADLINE_EXCEEDED: CallOptions deadline exceeded after 4.998s. Name resolution delay 0.000000000 seconds. [closed=[], open=[[connecting_and_lb_delay=4999ms, was_still_waiting]]] ``` ``` io.grpc.StatusException: UNKNOWN at io.grpc.xds.internal.security.DynamicSslContextProvider.onError(DynamicSslContextProvider.java:145) at io.grpc.xds.internal.security.certprovider.CertificateProvider$DistributorWatcher.onError(CertificateProvider.java:128) at io.grpc.xds.internal.security.certprovider.FileWatcherCertificateProvider.checkAndReloadCertificates(FileWatcherCertificateProvider.java:149) Caused by: java.util.ConcurrentModificationException at java.base/java.util.HashMap$HashIterator.nextNode(HashMap.java:1606) at java.base/java.util.HashMap$KeyIterator.next(HashMap.java:1629) at io.grpc.xds.internal.security.certprovider.CertificateProvider$DistributorWatcher.updateSpiffeTrustMap(CertificateProvider.java:120) at io.grpc.xds.internal.security.certprovider.FileWatcherCertificateProvider.checkAndReloadCertificates(FileWatcherCertificateProvider.java:144) ``` ``` java.util.ConcurrentModificationException at java.util.HashMap$HashIterator.nextNode(HashMap.java:1606) at java.util.HashMap$KeyIterator.next(HashMap.java:1629) at CertificateProvider$DistributorWatcher.onError(CertificateProvider.java:127) at FileWatcherCertificateProvider.checkAndReloadCertificates(FileWatcherCertificateProvider.java:149) ``` --- .../certprovider/CertificateProvider.java | 126 +++++++----- ...tProviderClientSslContextProviderTest.java | 9 +- ...tProviderServerSslContextProviderTest.java | 9 +- .../CertificateProviderStoreTest.java | 10 +- .../certprovider/CertificateProviderTest.java | 192 ++++++++++++++++++ 5 files changed, 273 insertions(+), 73 deletions(-) create mode 100644 xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderTest.java diff --git a/xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java b/xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java index 009bb7bf566..4944372f4c5 100644 --- a/xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java +++ b/xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java @@ -19,15 +19,19 @@ import static com.google.common.base.Preconditions.checkNotNull; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableSet; +import com.google.errorprone.annotations.concurrent.GuardedBy; import io.grpc.Status; +import io.grpc.SynchronizationContext; import io.grpc.xds.internal.security.Closeable; import java.security.PrivateKey; import java.security.cert.X509Certificate; -import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.logging.Level; +import java.util.logging.Logger; /** * A plug-in that provides certificates required by the xDS security component and created @@ -40,6 +44,8 @@ */ public abstract class CertificateProvider implements Closeable { + private static final Logger logger = Logger.getLogger(CertificateProvider.class.getName()); + /** A watcher is registered to receive certificate updates. */ public interface Watcher { void updateCertificate(PrivateKey key, List certChain); @@ -53,95 +59,111 @@ public interface Watcher { @VisibleForTesting public static final class DistributorWatcher implements Watcher { - private PrivateKey privateKey; - private List certChain; - private List trustedRoots; - private Map> spiffeTrustMap; - - @VisibleForTesting - final Set downstreamWatchers = new HashSet<>(); - - synchronized void addWatcher(Watcher watcher) { - downstreamWatchers.add(watcher); - if (privateKey != null && certChain != null) { - sendLastCertificateUpdate(watcher); - } - if (trustedRoots != null) { - sendLastTrustedRootsUpdate(watcher); - } - if (spiffeTrustMap != null) { - sendLastSpiffeTrustMapUpdate(watcher); + // All watcher callbacks must run in syncContext to avoid possibility of deadlock + private final SynchronizationContext syncContext = new SynchronizationContext( + new Thread.UncaughtExceptionHandler() { + @Override + public void uncaughtException(Thread t, Throwable e) { + logger.log(Level.SEVERE, "Uncaught exception in DistributorWatcher callback", e); + } + }); + + @GuardedBy("this") private PrivateKey privateKey; + @GuardedBy("this") private List certChain; + @GuardedBy("this") private List trustedRoots; + @GuardedBy("this") private Map> spiffeTrustMap; + + @GuardedBy("this") + private final Set downstreamWatchers = new HashSet<>(); + + void addWatcher(Watcher watcher) { + synchronized (this) { + downstreamWatchers.add(watcher); + if (privateKey != null && certChain != null) { + PrivateKey key = privateKey; + List chain = certChain; + syncContext.executeLater(() -> watcher.updateCertificate(key, chain)); + } + if (trustedRoots != null) { + List roots = trustedRoots; + syncContext.executeLater(() -> watcher.updateTrustedRoots(roots)); + } + if (spiffeTrustMap != null) { + Map> map = spiffeTrustMap; + syncContext.executeLater(() -> watcher.updateSpiffeTrustMap(map)); + } } + syncContext.drain(); } synchronized void removeWatcher(Watcher watcher) { downstreamWatchers.remove(watcher); } - @VisibleForTesting public Set getDownstreamWatchers() { - return Collections.unmodifiableSet(downstreamWatchers); - } - - private void sendLastCertificateUpdate(Watcher watcher) { - watcher.updateCertificate(privateKey, certChain); - } - - private void sendLastTrustedRootsUpdate(Watcher watcher) { - watcher.updateTrustedRoots(trustedRoots); - } - - private void sendLastSpiffeTrustMapUpdate(Watcher watcher) { - watcher.updateSpiffeTrustMap(spiffeTrustMap); + @VisibleForTesting public synchronized Set getDownstreamWatchers() { + return ImmutableSet.copyOf(downstreamWatchers); } @Override - public synchronized void updateCertificate(PrivateKey key, List certChain) { + public void updateCertificate(PrivateKey key, List certChain) { checkNotNull(key, "key"); checkNotNull(certChain, "certChain"); - privateKey = key; - this.certChain = certChain; - for (Watcher watcher : downstreamWatchers) { - sendLastCertificateUpdate(watcher); + synchronized (this) { + privateKey = key; + this.certChain = certChain; + for (Watcher watcher : downstreamWatchers) { + syncContext.executeLater(() -> watcher.updateCertificate(key, certChain)); + } } + syncContext.drain(); } @Override - public synchronized void updateTrustedRoots(List trustedRoots) { + public void updateTrustedRoots(List trustedRoots) { checkNotNull(trustedRoots, "trustedRoots"); - this.trustedRoots = trustedRoots; - for (Watcher watcher : downstreamWatchers) { - sendLastTrustedRootsUpdate(watcher); + synchronized (this) { + this.trustedRoots = trustedRoots; + for (Watcher watcher : downstreamWatchers) { + syncContext.executeLater(() -> watcher.updateTrustedRoots(trustedRoots)); + } } + syncContext.drain(); } @Override public void updateSpiffeTrustMap(Map> spiffeTrustMap) { - this.spiffeTrustMap = spiffeTrustMap; - for (Watcher watcher : downstreamWatchers) { - sendLastSpiffeTrustMapUpdate(watcher); + synchronized (this) { + this.spiffeTrustMap = spiffeTrustMap; + for (Watcher watcher : downstreamWatchers) { + syncContext.executeLater(() -> watcher.updateSpiffeTrustMap(spiffeTrustMap)); + } } + syncContext.drain(); } @Override - public synchronized void onError(Status errorStatus) { - for (Watcher watcher : downstreamWatchers) { - watcher.onError(errorStatus); + public void onError(Status errorStatus) { + synchronized (this) { + for (Watcher watcher : downstreamWatchers) { + syncContext.executeLater(() -> watcher.onError(errorStatus)); + } } + syncContext.drain(); } - X509Certificate getLastIdentityCert() { + synchronized X509Certificate getLastIdentityCert() { if (certChain != null && !certChain.isEmpty()) { return certChain.get(0); } return null; } - void close() { + synchronized void close() { downstreamWatchers.clear(); clearValues(); } - void clearValues() { + synchronized void clearValues() { privateKey = null; certChain = null; trustedRoots = null; @@ -152,7 +174,7 @@ void clearValues() { * Concrete subclasses will call this to register the {@link Watcher}. * * @param watcher to register - * @param notifyCertUpdates if true, the provider is required to call the watcher’s + * @param notifyCertUpdates if true, the provider is required to call the watcher's * updateCertificate method. Implies the Provider is capable of minting certificates. * Used by server-side and mTLS client-side. Note the Provider is always required * to call updateTrustedRoots to provide trusted-root updates. diff --git a/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderClientSslContextProviderTest.java b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderClientSslContextProviderTest.java index 37aa2c65b08..3cd8cd4ca8a 100644 --- a/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderClientSslContextProviderTest.java +++ b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderClientSslContextProviderTest.java @@ -421,14 +421,7 @@ public void testProviderForClient_sslContextException_onError() throws Exception TestCallback testCallback = new TestCallback(MoreExecutors.directExecutor()); provider.addCallback(testCallback); - try { - watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE))); - fail("exception expected"); - } catch (RuntimeException expected) { - assertThat(expected) - .hasMessageThat() - .contains("only static certificateValidationContext expected"); - } + watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE))); assertThat(testCallback.updatedThrowable).isNotNull(); assertThat(testCallback.updatedThrowable) .hasCauseThat() diff --git a/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderServerSslContextProviderTest.java b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderServerSslContextProviderTest.java index 93559f47245..e0d4af1a411 100644 --- a/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderServerSslContextProviderTest.java +++ b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderServerSslContextProviderTest.java @@ -339,14 +339,7 @@ public void testProviderForServer_sslContextException_onError() throws Exception TestCallback testCallback = new TestCallback(MoreExecutors.directExecutor()); provider.addCallback(testCallback); - try { - watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE))); - fail("exception expected"); - } catch (RuntimeException expected) { - assertThat(expected) - .hasMessageThat() - .contains("only static certificateValidationContext expected"); - } + watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE))); assertThat(testCallback.updatedThrowable).isNotNull(); assertThat(testCallback.updatedThrowable) .hasCauseThat() diff --git a/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderStoreTest.java b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderStoreTest.java index c0bc095eab6..fc9026e6fb6 100644 --- a/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderStoreTest.java +++ b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderStoreTest.java @@ -138,7 +138,7 @@ public void onePluginSameConfig_sameInstance() { (TestCertificateProvider) handle1.certProvider; assertThat(testCertificateProvider.startCalled).isEqualTo(1); CertificateProvider.DistributorWatcher distWatcher = testCertificateProvider.getWatcher(); - assertThat(distWatcher.downstreamWatchers).hasSize(2); + assertThat(distWatcher.getDownstreamWatchers()).hasSize(2); PrivateKey testKey = mock(PrivateKey.class); X509Certificate cert = mock(X509Certificate.class); List testList = ImmutableList.of(cert); @@ -154,7 +154,7 @@ public void onePluginSameConfig_sameInstance() { reset(mockWatcher2); handle1.close(); assertThat(testCertificateProvider.closeCalled).isEqualTo(0); - assertThat(distWatcher.downstreamWatchers).hasSize(1); + assertThat(distWatcher.getDownstreamWatchers()).hasSize(1); testCertificateProvider.getWatcher().updateCertificate(testKey, testList); verify(mockWatcher1, never()) .updateCertificate(any(PrivateKey.class), ArgumentMatchers.anyList()); @@ -189,7 +189,7 @@ public void onePluginSameConfig_secondWatcherAfterFirstNotify() { CertificateProvider.Watcher mockWatcher2 = mock(CertificateProvider.Watcher.class); CertificateProviderStore.Handle unused = certificateProviderStore.createOrGetProvider( "cert-name1", "plugin1", "config", mockWatcher2, true); - assertThat(distWatcher.downstreamWatchers).hasSize(2); + assertThat(distWatcher.getDownstreamWatchers()).hasSize(2); // updates sent to the second watcher verify(mockWatcher2, times(1)).updateCertificate(eq(testKey), eq(testList)); verify(mockWatcher2, times(1)).updateTrustedRoots(eq(testList)); @@ -290,9 +290,9 @@ private static void checkDifferentInstances( assertThat(testCertificateProvider2.certProviderProvider) .isSameInstanceAs(certProviderProvider2); CertificateProvider.DistributorWatcher distWatcher1 = testCertificateProvider1.getWatcher(); - assertThat(distWatcher1.downstreamWatchers).hasSize(1); + assertThat(distWatcher1.getDownstreamWatchers()).hasSize(1); CertificateProvider.DistributorWatcher distWatcher2 = testCertificateProvider2.getWatcher(); - assertThat(distWatcher2.downstreamWatchers).hasSize(1); + assertThat(distWatcher2.getDownstreamWatchers()).hasSize(1); PrivateKey testKey1 = mock(PrivateKey.class); X509Certificate cert1 = mock(X509Certificate.class); List testList1 = ImmutableList.of(cert1); diff --git a/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderTest.java b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderTest.java new file mode 100644 index 00000000000..8dcf30100d8 --- /dev/null +++ b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderTest.java @@ -0,0 +1,192 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.grpc.xds.internal.security.certprovider; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import io.grpc.Status; +import io.grpc.xds.internal.security.certprovider.CertificateProvider.DistributorWatcher; +import io.grpc.xds.internal.security.certprovider.CertificateProvider.Watcher; +import java.security.PrivateKey; +import java.security.cert.X509Certificate; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests notification delivery when callbacks change certificate provider subscriptions. */ +@RunWith(JUnit4.class) +public class CertificateProviderTest { + private final DistributorWatcher distributor = new DistributorWatcher(); + private final Watcher watcher1 = mock(Watcher.class); + private final Watcher watcher2 = mock(Watcher.class); + private final Map> trustMap = + ImmutableMap.of("example.org", ImmutableList.of(mock(X509Certificate.class))); + + @Test + public void updateSpiffeTrustMap_doesNotOverlapCertificateReplay() throws Exception { + PrivateKey key = mock(PrivateKey.class); + List certChain = ImmutableList.of(mock(X509Certificate.class)); + distributor.updateCertificate(key, certChain); + CountDownLatch replayStarted = new CountDownLatch(1); + CountDownLatch finishReplay = new CountDownLatch(1); + CountDownLatch updateStarted = new CountDownLatch(1); + AtomicBoolean replayingCertificate = new AtomicBoolean(); + AtomicBoolean overlappingNotifications = new AtomicBoolean(); + doAnswer(invocation -> { + replayingCertificate.set(true); + replayStarted.countDown(); + assertThat(finishReplay.await(10, TimeUnit.SECONDS)).isTrue(); + replayingCertificate.set(false); + return null; + }).when(watcher1).updateCertificate(key, certChain); + doAnswer(invocation -> { + overlappingNotifications.compareAndSet(false, replayingCertificate.get()); + return null; + }).when(watcher1).updateSpiffeTrustMap(trustMap); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future registration = executor.submit(() -> distributor.addWatcher(watcher1)); + assertThat(replayStarted.await(10, TimeUnit.SECONDS)).isTrue(); + Future update = executor.submit(() -> { + updateStarted.countDown(); + distributor.updateSpiffeTrustMap(trustMap); + }); + assertThat(updateStarted.await(10, TimeUnit.SECONDS)).isTrue(); + try { + // A serialized update waits for registration to finish replaying the certificate. + update.get(1, TimeUnit.SECONDS); + } catch (TimeoutException expected) { + // Release registration below so both tasks can finish before checking delivery. + } finally { + finishReplay.countDown(); + } + registration.get(10, TimeUnit.SECONDS); + update.get(10, TimeUnit.SECONDS); + + assertThat(overlappingNotifications.get()).isFalse(); + verify(watcher1).updateSpiffeTrustMap(trustMap); + } finally { + finishReplay.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + public void updateCertificate_watcherRemovedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + PrivateKey key = mock(PrivateKey.class); + List certChain = ImmutableList.of(mock(X509Certificate.class)); + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.removeWatcher(firstWatcher); + return null; + }).when(firstWatcher).updateCertificate(key, certChain); + + distributor.updateCertificate(key, certChain); + + verify(watcher1).updateCertificate(key, certChain); + verify(watcher2).updateCertificate(key, certChain); + } + + @Test + public void updateTrustedRoots_watcherRemovedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + List trustedRoots = ImmutableList.of(mock(X509Certificate.class)); + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.removeWatcher(firstWatcher); + return null; + }).when(firstWatcher).updateTrustedRoots(trustedRoots); + + distributor.updateTrustedRoots(trustedRoots); + + verify(watcher1).updateTrustedRoots(trustedRoots); + verify(watcher2).updateTrustedRoots(trustedRoots); + } + + @Test + public void updateSpiffeTrustMap_watcherAddedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + Watcher addedWatcher = mock(Watcher.class); + // Select the first callback without depending on HashSet's iteration order. At least one + // existing watcher still needs its initial trust map when this callback changes the set. + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.addWatcher(addedWatcher); + return null; + }).when(firstWatcher).updateSpiffeTrustMap(trustMap); + + distributor.updateSpiffeTrustMap(trustMap); + + verify(watcher1).updateSpiffeTrustMap(trustMap); + verify(watcher2).updateSpiffeTrustMap(trustMap); + verify(addedWatcher).updateSpiffeTrustMap(trustMap); + } + + @Test + public void updateSpiffeTrustMap_watcherRemovedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.removeWatcher(firstWatcher); + return null; + }).when(firstWatcher).updateSpiffeTrustMap(trustMap); + + distributor.updateSpiffeTrustMap(trustMap); + + verify(watcher1).updateSpiffeTrustMap(trustMap); + verify(watcher2).updateSpiffeTrustMap(trustMap); + assertThat(distributor.getDownstreamWatchers()).doesNotContain(firstWatcher); + } + + @Test + public void onError_watcherRemovedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.removeWatcher(firstWatcher); + return null; + }).when(firstWatcher).onError(Status.UNAVAILABLE); + + distributor.onError(Status.UNAVAILABLE); + + verify(watcher1).onError(Status.UNAVAILABLE); + verify(watcher2).onError(Status.UNAVAILABLE); + assertThat(distributor.getDownstreamWatchers()).doesNotContain(firstWatcher); + } +}