diff --git a/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java b/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java index d3384cbbe4e..4c6823f844a 100644 --- a/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java +++ b/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java @@ -676,6 +676,8 @@ private final class ResourceSubscriber { private ResourceMetadata metadata; @Nullable private String errorDescription; + @Nullable + private Status lastError; ResourceSubscriber(XdsResourceType type, String resource) { syncContext.throwIfNotInThisSynchronizationContext(); @@ -712,11 +714,16 @@ void addWatcher(ResourceWatcher watcher, Executor watcherExecutor) { watchers.put(watcher, watcherExecutor); T savedData = data; boolean savedAbsent = absent; + Status savedError = lastError; watcherExecutor.execute(() -> { if (errorDescription != null) { watcher.onError(Status.INVALID_ARGUMENT.withDescription(errorDescription)); return; } + if (savedError != null) { + watcher.onError(savedError); + return; + } if (savedData != null) { notifyWatcher(watcher, savedData); } else if (savedAbsent) { @@ -808,6 +815,7 @@ void onData(ParsedResource parsedResource, String version, long updateTime, this.metadata = ResourceMetadata .newResourceMetadataAcked(parsedResource.getRawResource(), version, updateTime); absent = false; + lastError = null; if (resourceDeletionIgnored) { logger.log(XdsLogLevel.FORCE_INFO, "xds server {0}: server returned new version " + "of resource for which we previously ignored a deletion: type {1} name {2}", @@ -857,6 +865,7 @@ void onAbsent(@Nullable ProcessingTracker processingTracker, ServerInfo serverIn if (!absent) { data = null; absent = true; + lastError = null; metadata = serverInfo.resourceTimerIsTransientError() ? ResourceMetadata.newResourceMetadataTimeout() : ResourceMetadata.newResourceMetadataDoesNotExist(); @@ -894,6 +903,7 @@ void onError(Status error, @Nullable ProcessingTracker tracker) { Status errorAugmented = Status.fromCode(error.getCode()) .withDescription(description + "nodeID: " + bootstrapInfo.node().getId()) .withCause(error.getCause()); + this.lastError = errorAugmented; for (ResourceWatcher watcher : watchers.keySet()) { if (tracker != null) { diff --git a/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java b/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java index 89668485b96..19266b0d289 100644 --- a/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java +++ b/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java @@ -285,6 +285,8 @@ public long currentTimeNanos() { @Mock private ResourceWatcher ldsResourceWatcher; @Mock + private ResourceWatcher ldsResourceWatcher2; + @Mock private ResourceWatcher rdsResourceWatcher; @Mock private ResourceWatcher cdsResourceWatcher; @@ -694,6 +696,24 @@ public void ldsResourceUpdated_withXdstpResourceName_withUnknownAuthority() { assertThat(resourceDiscoveryCalls.poll()).isNull(); } + @Test + public void ldsResource_onError_cachedForNewWatcher() { + xdsClient.watchXdsResource(XdsListenerResource.getInstance(), LDS_RESOURCE, + ldsResourceWatcher); + DiscoveryRpcCall call = resourceDiscoveryCalls.poll(); + call.sendCompleted(); + verify(ldsResourceWatcher).onError(errorCaptor.capture()); + Status initialError = errorCaptor.getValue(); + xdsClient.watchXdsResource(XdsListenerResource.getInstance(), LDS_RESOURCE, + ldsResourceWatcher2); + ArgumentCaptor secondErrorCaptor = ArgumentCaptor.forClass(Status.class); + verify(ldsResourceWatcher2).onError(secondErrorCaptor.capture()); + Status cachedError = secondErrorCaptor.getValue(); + + assertThat(cachedError).isEqualTo(initialError); + assertThat(resourceDiscoveryCalls.poll()).isNull(); + } + @Test public void ldsResponseErrorHandling_allResourcesFailedUnpack() { DiscoveryRpcCall call = startResourceWatcher(XdsListenerResource.getInstance(), LDS_RESOURCE,