xds: Make RawMessageClientInterceptor conditional on ext_proc flags (v1.83.x backport) (#13018)

Backport of #13015 to v1.83.x.
---
Only add RawMessageClientInterceptor to the interceptor chain in
XdsNameResolver when either GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT or
GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER is true. This prevents
RawMessageClientInterceptor from corrupting method descriptors and
causing empty payloads on retry attempts for regular xDS channels.

Temporary fix for #13010

Author: kannanjgithub

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

@@ -905,7 +905,10 @@ private ClientInterceptor createFilters(
       }
 
       ImmutableList.Builder<ClientInterceptor> withRawMessage = ImmutableList.builder();
-      withRawMessage.add(new RawMessageClientInterceptor());
+      if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)
+          || GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", false)) {
+        withRawMessage.add(new RawMessageClientInterceptor());
+      }
       withRawMessage.addAll(filterInterceptors.build());
       return combineInterceptors(withRawMessage.build());
     }

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

@@ -2934,7 +2934,7 @@ private final class TestChannel extends Channel {
     @Override
     public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
         MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
-      TestCall<ReqT, RespT> call = new TestCall<>(callOptions);
+      TestCall<ReqT, RespT> call = new TestCall<>(methodDescriptor, callOptions);
       testCall = call;
       return call;
     }
@@ -2946,11 +2946,13 @@ public String authority() {
   }
 
   private static final class TestCall<ReqT, RespT> extends NoopClientCall<ReqT, RespT> {
+    final MethodDescriptor<ReqT, RespT> methodDescriptor;
     // CallOptions actually received from the channel when the call is created.
     final CallOptions callOptions;
     ClientCall.Listener<RespT> listener;
 
-    TestCall(CallOptions callOptions) {
+    TestCall(MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
+      this.methodDescriptor = methodDescriptor;
       this.callOptions = callOptions;
     }
 
@@ -3038,4 +3040,81 @@ public void onMessage(String message) {
         channel, METHOD_SAY_HELLO, CallOptions.DEFAULT, "World");
     assertThat(response).isEqualTo("Hello World");
   }
+
+  @Test
+  public void rawMessageClientInterceptor_flagFalse() {
+    String origClientProp = System.getProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT");
+    String origServerProp = System.getProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER");
+    System.setProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", "false");
+    System.setProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", "false");
+    try {
+      filterStateTestSetupResolver();
+      FakeXdsClient xdsClient = (FakeXdsClient) resolver.getXdsClient();
+      VirtualHost vhost = filterStateTestVhost();
+
+      xdsClient.deliverLdsUpdateWithFilters(vhost, filterStateTestConfigs(STATEFUL_1));
+      createAndDeliverClusterUpdates(xdsClient, cluster1);
+
+      // When flags are false, RawMessageClientInterceptor is not added.
+      assertClusterResolutionResult(call1, cluster1);
+      assertThat(testCall.methodDescriptor).isSameInstanceAs(call1.methodDescriptor);
+    } finally {
+      restoreProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", origClientProp);
+      restoreProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", origServerProp);
+    }
+  }
+
+  @Test
+  public void rawMessageClientInterceptor_flagTrue() {
+    String origClientProp = System.getProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT");
+    String origServerProp = System.getProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER");
+
+    // When GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT is true, RawMessageClientInterceptor is added.
+    System.setProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", "true");
+    System.setProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", "false");
+    try {
+      filterStateTestSetupResolver();
+      FakeXdsClient xdsClient = (FakeXdsClient) resolver.getXdsClient();
+      VirtualHost vhost = filterStateTestVhost();
+
+      xdsClient.deliverLdsUpdateWithFilters(vhost, filterStateTestConfigs(STATEFUL_1));
+      createAndDeliverClusterUpdates(xdsClient, cluster1);
+
+      assertClusterResolutionResult(call1, cluster1);
+      assertThat(testCall.methodDescriptor).isNotSameInstanceAs(call1.methodDescriptor);
+    } finally {
+      restoreProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", origClientProp);
+      restoreProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", origServerProp);
+    }
+
+    resolver.shutdown();
+    reset(mockListener);
+    when(mockListener.onResult2(any())).thenReturn(Status.OK);
+
+    // When GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER is true, RawMessageClientInterceptor is added.
+    System.setProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", "false");
+    System.setProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", "true");
+    try {
+      filterStateTestSetupResolver();
+      FakeXdsClient xdsClient = (FakeXdsClient) resolver.getXdsClient();
+      VirtualHost vhost = filterStateTestVhost();
+
+      xdsClient.deliverLdsUpdateWithFilters(vhost, filterStateTestConfigs(STATEFUL_1));
+      createAndDeliverClusterUpdates(xdsClient, cluster1);
+
+      assertClusterResolutionResult(call1, cluster1);
+      assertThat(testCall.methodDescriptor).isNotSameInstanceAs(call1.methodDescriptor);
+    } finally {
+      restoreProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", origClientProp);
+      restoreProperty("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", origServerProp);
+    }
+  }
+
+  private static void restoreProperty(String key, @Nullable String value) {
+    if (value == null) {
+      System.clearProperty(key);
+    } else {
+      System.setProperty(key, value);
+    }
+  }
 }