xds: Changes to XdsClient Watcher APIs #12446

shivaspeaks · 2025-10-27T14:23:14Z

Implements https://github.com/grpc/proposal/blob/master/A88-xds-data-error-handling.md#a88-xds-data-error-handling

kannanjgithub

Sending what I could review today.

kannanjgithub · 2025-10-28T12:56:55Z

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

+     * Called to deliver a transient error that should not affect the watcher's use of any
+     * previously received resource.
     *
     * <p>Note that we expect that the implementer to:


I think this comment now applies to both methods since both methods take Status argument now.

We can remove it from the method's Javadoc as you have put in the class' Javadoc.

kannanjgithub · 2025-10-28T13:12:33Z

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

              type, resource);
-          respTimer = null;
          onAbsent(null, activeCpc.getServerInfo());
+          respTimer = null;


Why swap the order of these statements?

The original code was setting respTimer = null before calling onAbsent(), which caused the timeout context to be lost, resulting in a generic "does not exist" error.

The original for onAbsent method did not have a check for if (respTimer == null). It only looked at the serverInfo.resourceTimerIsTransientError() flag to decide between calling watcher.onError() or watcher.onResourceDoesNotExist(). It could not distinguish between a timeout and other "not found" conditions.

kannanjgithub · 2025-10-28T13:13:19Z

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

          logger.log(XdsLogLevel.WARNING, "No working fallback XDS Servers found from {0}",
-              activeCpClient.getServerInfo().target());
+              activeCpc.getServerInfo().target());
        }


The changes above here seem unnecessary?

kannanjgithub · 2025-10-28T13:38:37Z

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

+          executor.execute(() -> {
            try {
-              notifyWatcher(watcher, data);
+              watcher.onResourceChanged(update); // Call the new method


Remove the comment as we are passing StatusOr argument.

kannanjgithub · 2025-10-28T14:09:18Z

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

+            continue;
+          }
+          if (subscriber.hasResult()) {
+            subscriber.onError(status, null); // This will become an onAmbientError


As the handling to call either onDataChanged or onAmbientError happens in ResourceSubscriber.onError, why not simply call it? You are only calling if subscriber.hasResult() is true. What if when it is false? Previously the onError was getting called in that case but not now.

kannanjgithub

Review comments.

kannanjgithub · 2025-11-17T12:41:51Z

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

+     * Called to deliver a transient error that should not affect the watcher's use of any
+     * previously received resource.
     *
     * <p>Note that we expect that the implementer to:


We can remove it from the method's Javadoc as you have put in the class' Javadoc.

kannanjgithub · 2025-11-17T12:54:00Z

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

      if (!Objects.equals(oldData, data)) {
-        for (ResourceWatcher<T> watcher : watchers.keySet()) {
+        StatusOr<T> update = StatusOr.fromValue(data);
+        for (Map.Entry<ResourceWatcher<T>, Executor> entry : watchers.entrySet()) {


Why change from iterating the keySet to iterating the entrySet? There is nothing wrong with it but it seems like an unnecessary change.
Can keep the executor inlined like before as well.

Yeah, I did that intentionally because my intention for changing to entrySet was to avoid the extra lookup for each key to get executor below. But yeah this change is unnecessary, I'll revert back.

kannanjgithub · 2025-11-17T13:49:57Z

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

+
+        Status status;
+        if (respTimer == null) {
+          status = Status.NOT_FOUND.withDescription("Resource " + resource + " does not exist");


For resource deletions, if fail_on_data_error is false, we should not be setting data = null; and should be calling onAmbientError(NOT_FOUND). lastError should also be set to NOT_FOUND so both the data and the last error are available for replay for new watchers.
Only for fail_on_data_error true we should drop the resource (data = null;) and call onResourceChanged.

fail_on_data_error is a server feature that needs to be added. In this PR we are focusing only on Changes to XdsClient Watcher APIs as stated in the PR description as well.

kannanjgithub · 2025-11-17T13:56:00Z

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

      for (Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscriberMap :
          resourceSubscribers.values()) {
        for (ResourceSubscriber<? extends ResourceUpdate> subscriber : subscriberMap.values()) {
-          if (subscriber.hasResult() || !authoritiesForClosedCpc.contains(subscriber.authority)) {


Why remove subscriber.hasResult()?

The if (subscriber.hasResult()) check prevented subscribers without a cached resource from being notified of stream errors. By removing it, we now correctly call onError for all subscribers, and the onError method itself handles the logic of whether to call onResourceChanged or onAmbientError.

kannanjgithub · 2025-11-17T13:56:44Z

xds/src/main/java/io/grpc/xds/XdsDependencyManager.java

    private StatusOr<T> data;
+    @Nullable
+    @SuppressWarnings("unused")
+    private Status ambientError;


Will there be a separate PR for propagating this to the data plane rpc error?

kannanjgithub · 2025-11-17T13:59:45Z

xds/src/main/java/io/grpc/xds/XdsServerWrapper.java

      }
+
      if (!pendingRds.isEmpty()) {
-        // filter chain state has not yet been applied to filterChainSelectorManager and there


Restore this comment.

shivaspeaks · 2025-11-21T08:16:41Z

Note: The failing test log shown here in github Actions is not failing in local even after clean run/build.

kannanjgithub · 2025-11-24T06:07:19Z

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

      }

-      Status newStatus = status;
-      if (responseReceived) {


Are these changes from grfc A88?

The previous logic would incorrectly treat any stream error as Status.OK if a response had already been received. This would have prevented us from ever calling the new onAmbientError callback for these transient failures.

This new implementation ensures that all stream failures are correctly propagated, which then allows XdsClientImpl to decide whether to call onResourceChanged or onAmbientError as defined in gRFC A88.

(Continuing reply here, previous reply sent from email landed on the main thread.)

The question is whether the definition of a transient error has changed w.r.t. a resource already received on the ADS stream or not. C-core also ignores error statuses when a resource has already been received and this results in watchers getting error notifications only if no resource has been received.

And since the XdsClient restarts the ADS stream upon termination, the next occurrence of the error status will get handled as an error on the restarted stream.

kannanjgithub · 2025-11-24T06:28:09Z

Why is it incorrect? A88 does not say it is modifying the A57 ADS stream error handling behavior after a response is received, A88 only mentions cases where no response was received on the ADS stream and the stream was terminated.

…

On Mon, Nov 24, 2025 at 11:44 AM MV Shiva ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java <#12446 (comment)>: > @@ -453,44 +453,29 @@ private void handleRpcStreamClosed(Status status) { stopwatch.reset(); } - Status newStatus = status; - if (responseReceived) { The previous logic would incorrectly treat any stream error as Status.OK if a response had already been received. This would have prevented us from ever calling the new onAmbientError callback for these transient failures. This new implementation ensures that all stream failures are correctly propagated, which then allows XdsClientImpl to decide whether to call onResourceChanged or onAmbientError as defined in gRFC A88. — Reply to this email directly, view it on GitHub <#12446 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AHB2YHA3EQ6BWCHDQ3P2LMT36KO4VAVCNFSM6AAAAACKKR4ZPSVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZTIOJYGQ3DGNJUGE> . You are receiving this because your review was requested.Message ID: ***@***.***>

shivaspeaks · 2025-11-24T07:33:16Z

The purpose of gRFC A88 is to introduce onAmbientError to notify watchers of transient errors without them having to drop their cached data. If the underlying ControlPlaneClient hides the error and reports Status.OK, then XdsClientImpl would never know a transient failure occurred and would therefore never be able to call onAmbientError on the watchers. This would completely break this new mechanism.

kannanjgithub · 2025-11-24T06:09:39Z

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


    ControlPlaneClient activeCpClient = getActiveCpc(subscriber.authority);
    if (cpcToUse != activeCpClient) {
-      addCpcToAuthority(subscriber.authority, cpcToUse); // makes it active


Add the comment back.

kannanjgithub · 2025-11-24T06:17:51Z

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

-          notifyWatcher(watcher, savedData);
+          watcher.onResourceChanged(StatusOr.fromValue(savedData));
+          if (savedError != null) {
+            watcher.onAmbientError(savedError);


This statement is not tested.

kannanjgithub · 2025-11-24T08:56:08Z

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

      }

-      Status newStatus = status;
-      if (responseReceived) {


(Continuing reply here, previous reply sent from email landed on the main thread.)

The question is whether the definition of a transient error has changed w.r.t. a resource already received on the ADS stream or not. C-core also ignores error statuses when a resource has already been received and this results in watchers getting error notifications only if no resource has been received.

And since the XdsClient restarts the ADS stream upon termination, the next occurrence of the error status will get handled as an error on the restarted stream.

shivaspeaks · 2025-11-24T11:15:10Z

Note: I added a TODO in connect_then_mainServerDown_fallbackServerUp in XdsClientFallbackTest. Unable to write this test. Any suggestions on this would be appreciated.

shivaspeaks · 2025-11-27T14:40:22Z

if (subscriber.hasResult() || !authoritiesForClosedCpc.contains(subscriber.authority)) {
    subscriber.onError(status, null);
    continue;
}

This above code block was the initial fix for the failing fallback test that was mentioned above.
But it introduced the flakiness in testTwoBadUrl() in XdsClientFallbackTest.java. The test became flaky because sometimes it might see the premature garbageUri1 error, and sometimes it might run long enough to also see the final garbageUri2 error, and ldsUpdateCaptor.getValue() would happen to hold the correct final value. The behavior was no longer deterministic.

The solution is to restructure the logic in the for loop to handle different subscriber states more clearly and deterministically. The updated logic is more clearer:

If a subscriber already has data, it should get an ambient error.
If a subscriber does not have data, the client should attempt to fall back.
Only if the subscriber has no data AND all fallback attempts have failed should a definitive error be reported.

shivaspeaks and others added 20 commits August 11, 2025 20:09

MAINTAINERS.md: Add Shiva

21352b7

Merge branch 'grpc:master' into master

458735d

Update XdsTrustManagerFactoryTest.java

4f064fa

refactor and add documentation

0f96110

Merge branch 'master' of https://github.com/grpc/grpc-java

d7ef02a

Merge branch 'grpc:master' into master

405217b

Merge branch 'grpc:master' into master

587969d

xds: client watcher API changes

b1d0d6d

xds: client watcher API changes

3fcff8d

xds: client watcher API changes

bf905c2

xds: client watcher API changes

4f5f3b3

xds: client watcher API changes

494a75a

xds: client watcher API changes

a5117ad

xds: client watcher API changes

352ec9e

xds: client watcher API changes

148f105

xds: client watcher API changes

0c78954

xds: client watcher API changes

b5571d1

watcher api changes

c3a43b2

watcher api changes

910b20b

Merge pull request #3 from shivaspeaks/client-watcher-changes-2

319c113

shivaspeaks requested review from ejona86 and kannanjgithub October 27, 2025 14:23

shivaspeaks added 3 commits October 27, 2025 21:39

watcher api changes

2baf738

Merge branch 'master' of https://github.com/shivaspeaks/grpc-java

a51ac28

Merge branch 'client-watcher-changes-2' of https://github.com/shivasp…

3779aea

…eaks/grpc-java

shivaspeaks mentioned this pull request Oct 28, 2025

xds: client watcher API update #12046

Closed

kannanjgithub reviewed Oct 28, 2025

View reviewed changes

address comments

4ba6de4

kannanjgithub reviewed Nov 17, 2025

View reviewed changes

address comments

1fb0ff2

shivaspeaks requested a review from kannanjgithub November 21, 2025 07:39

kannanjgithub reviewed Nov 24, 2025

View reviewed changes

address comments

f9c8af6

add unit test

aead6f9

shivaspeaks requested a review from kannanjgithub November 25, 2025 04:46

shivaspeaks added 2 commits November 27, 2025 18:25

fix fallback

2601886

Fix flakyness of testTwoBadUrl

9900277

kannanjgithub approved these changes Nov 28, 2025

View reviewed changes

xds: Changes to XdsClient Watcher APIs #12446

Are you sure you want to change the base?

xds: Changes to XdsClient Watcher APIs #12446

Uh oh!

Conversation

shivaspeaks commented Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kannanjgithub left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kannanjgithub left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shivaspeaks commented Nov 21, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kannanjgithub Nov 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kannanjgithub commented Nov 24, 2025 via email

Uh oh!

shivaspeaks commented Nov 24, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kannanjgithub Nov 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shivaspeaks commented Nov 24, 2025

Uh oh!

shivaspeaks commented Nov 27, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

shivaspeaks commented Oct 27, 2025 •

edited

Loading

kannanjgithub Nov 24, 2025 •

edited

Loading

kannanjgithub Nov 24, 2025 •

edited

Loading