xds: xdsClient caches transient error for new watchers (v1.75.x backport) (#12291)

Backport of #12262 to v1.75.x. --- Fixes #11672 Co-authored-by: MV Shiva <speakupshiva@gmail.com>

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java

@@ -676,6 +676,8 @@ private final class ResourceSubscriber<T extends ResourceUpdate> {
  private ResourceMetadata metadata;
  @Nullable
  private String errorDescription;
+ @Nullable
+ private Status lastError;
 
  ResourceSubscriber(XdsResourceType<T> type, String resource) {
  syncContext.throwIfNotInThisSynchronizationContext();
@@ -712,11 +714,16 @@ void addWatcher(ResourceWatcher<T> 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<T> 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<T> watcher : watchers.keySet()) {
  if (tracker != null) {

xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java

@@ -285,6 +285,8 @@ public long currentTimeNanos() {
  @Mock
  private ResourceWatcher<LdsUpdate> ldsResourceWatcher;
  @Mock
+ private ResourceWatcher<LdsUpdate> ldsResourceWatcher2;
+ @Mock
  private ResourceWatcher<RdsUpdate> rdsResourceWatcher;
  @Mock
  private ResourceWatcher<CdsUpdate> 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<Status> 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,