[kannanjgithub]

1. Supporting retries

Previously, RawMessageClientInterceptor.sendMessage() forwarded a single-use
InputStream directly to rawCall.sendMessage(). When retries were triggered by
RetriableStream, attempt 1 drained the InputStream to EOF. Subsequent retry
attempts would then serialize the already-drained stream, sending an empty (0-byte)
request payload.

Fix this by:

1. Buffering the request stream into an immutable ByteString (using Drainable
   when supported) in RawMessageClientInterceptor.sendMessage() and wrapping it
   in a KnownLengthInputStream.
2. Exposing getByteString() on KnownLengthInputStream.
3. Having RAW_MARSHALLER.stream() return a fresh KnownLengthInputStream wrapping
   the underlying ByteString on each attempt, allowing retries to read the payload
   from byte 0.

2. Conditional addition of RawMessageClientInterceptor

Also we need to only install RawMessageClientInterceptor when a filter requires payload access to avoid the performance hit for non ext_proc cases. As suggested in the grfc discussion, introduce Filter.requiresPayloadAccess(config, overrideConfig) to allow filters to declare whether they need access to message payloads. Update ExternalProcessorFilter to implement this method, returning false when body send mode is NONE for both request and response. Update XdsNameResolver to only install RawMessageClientInterceptor when at least one configured filter requires payload access. Generalize ExternalProcessorClientInterceptor to support generic request/response types when payload interception is not needed.

Generalizing ExternalProcessorClientInterceptor to Support Pass-through Types

ExternalProcessorClientInterceptor previously hardcoded ClientCall<InputStream, InputStream> and SimpleForwardingClientCallListener<InputStream>, assuming RawMessageClientInterceptor was always present earlier in the chain to convert application messages to InputStream.

Now that RawMessageClientInterceptor is omitted when body send mode is NONE, ExternalProcessorClientInterceptor directly handles the application's request and response types (ReqT, RespT). DataPlaneClientCall and DataPlaneListener were generalized to <ReqT, RespT> to avoid compiler-generated synthetic bridge casts to InputStream, which would otherwise result in a runtime ClassCastException when sending or receiving application messages.

Fixes #13010

[sauravzg]

What's the best practice on handling exceptions in sendMessage ? Should we cancel the call and close listener or something when an exception happens with this error?

[kannanjgithub]

I had thought the application can catch and cancel the call, but looking at how ClientCallImpl.sendMessageInternal handles exceptions by canceling the stream and the application gets notified via an onClose callback, I have now followed the same, and added a new unit test rawMessageClientInterceptor_sendMessageIoException_cancelsCall to cover this.

[sauravzg]

So, might be worth discussing, but since we trigger cancel on sendmessage, but we don't have a way to make the application aware of it, the application might send another message or halfclose after cancellation triggering illegalstateexception.

We could either add a cancellation boolean or replace rawcall with noopcall to prevent it (I believe CheckedForwardingClientCall does this) or do cancellation(cleanup) + throw(making application handle error).

[sauravzg]

I am not sure how, but we probably need to unit test the behavior when the RawInterceptor isn't installed. I think all existing tests install it unconditionally.

[kannanjgithub]

I have updated the helper ExternalProcessorClientInterceptorTest.interceptCall() to mirror XdsNameResolver.createFilters():

• Only install RawMessageClientInterceptor(interceptReq, interceptResp) when interceptReq || interceptResp is true.
• That way, all existing header/trailer-only tests in ExternalProcessorClientInterceptorTest will automatically run without RawMessageClientInterceptor installed (and one-way body tests will only intercept the configured direction).

[sauravzg]

Looks good at an initial glance, just make sure to verify with jacoco reports on coverage numbers.

[sauravzg]

Approving this PR. But while reviewing this PR I found some other unrelated pre-existing bugs in our ext proc implementation which we should discuss offline and fix.