xds: Send protocol config and request attributes on first message to ext-proc

- Centralize decoration of ProcessingRequest messages within sendToExtProc.
- Ensure ProtocolConfiguration is populated on the very first message dispatched to the external processor.
- Ensure request attributes are collected and attached to the first client-to-server message (either request headers or request body).
- Add Category 3 tests to verify protocol configuration propagation across different processing mode setups.
- Add Category 4 tests to verify request attributes propagation on headers, body, or omitted correctly.

Author: kannanjgithub

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

@@ -442,6 +442,9 @@ private enum EventType {
     private long serverHeadersStartNanos;
     private long serverTrailersStartNanos;
 
+    private boolean protocolConfigSent = false;
+    private ImmutableMap<String, Struct> collectedAttributes;
+    private boolean requestAttributesSent = false;
     private volatile Metadata requestHeaders;
     final AtomicReference<DataPlaneCallState> dataPlaneCallState =
         new AtomicReference<>(DataPlaneCallState.IDLE);
@@ -795,6 +798,9 @@ public void onCompleted() {
         }
       });
 
+      this.collectedAttributes = collectAttributes(
+          config.getRequestAttributes(), method, channel, headers);
+
       boolean sendRequestHeaders =
           currentProcessingMode.getRequestHeaderMode() == ProcessingMode.HeaderSendMode.SEND
           || currentProcessingMode.getRequestHeaderMode()
@@ -806,12 +812,6 @@ public void onCompleted() {
                 .setHeaders(toHeaderMap(headers, config.getForwardRulesConfig()))
                 .setEndOfStream(false)
                 .build())
-            .putAllAttributes(
-                collectAttributes(config.getRequestAttributes(), method, channel, headers))
-            .setProtocolConfig(ProtocolConfiguration.newBuilder()
-                .setRequestBodyMode(currentProcessingMode.getRequestBodyMode())
-                .setResponseBodyMode(currentProcessingMode.getResponseBodyMode())
-                .build())
             .build());
       }
 
@@ -835,8 +835,30 @@ private void sendToExtProc(ProcessingRequest request) {
         }
 
         ProcessingRequest requestToSend = request;
+        if (!protocolConfigSent) {
+          requestToSend = ProcessingRequest.newBuilder(requestToSend)
+              .setProtocolConfig(ProtocolConfiguration.newBuilder()
+                  .setRequestBodyMode(currentProcessingMode.getRequestBodyMode())
+                  .setResponseBodyMode(currentProcessingMode.getResponseBodyMode())
+                  .build())
+              .build();
+          protocolConfigSent = true;
+        }
+
+        boolean isClientServerMessage =
+            requestToSend.hasRequestHeaders() || requestToSend.hasRequestBody();
+        if (isClientServerMessage
+            && !requestAttributesSent
+            && collectedAttributes != null
+            && !collectedAttributes.isEmpty()) {
+          requestToSend = ProcessingRequest.newBuilder(requestToSend)
+              .putAllAttributes(collectedAttributes)
+              .build();
+          requestAttributesSent = true;
+        }
+
         if (config.getObservabilityMode()) {
-          requestToSend = ProcessingRequest.newBuilder(request)
+          requestToSend = ProcessingRequest.newBuilder(requestToSend)
               .setObservabilityMode(true)
               .build();
         }