#20413

Committed 20 Aug 2026 05:04AM UTC coverage: 89.2% (+0.006%) from 89.194%

Build # #20413

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Prevent concurrent cancellations of data plane call in ExternalProcessorClientInterceptor (#12996)

If the ext-proc stream terminates with a non-OK status, the interceptor
cancels the downstream data plane call (on the executor thread).
Concurrently, the application thread may call cancel() on the returned
proxy call (e.g. for user cancellation or cleanup).

Since ClientCallImpl does not synchronize its cancelCalled field,
concurrent cancellations from these two threads resulted in a TSAN data
race (that was discussed in
https://github.com/grpc/grpc-java/pull/12975).

This commit introduces an AtomicBoolean downstreamCancelled in
DataPlaneClientCall to guard all downstream cancellations, ensuring that
only the first cancellation is forwarded to delayedCall/super.cancel().
This deduplicates and serializes cancellation handling from both the
application and the interceptor threads, preventing the data race.

Coverage Stats

38554 of 43222 relevant lines covered (89.2%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.14

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

/*
 * Copyright 2024 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.xds;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.applyHeaderMutations;
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.collectAttributes;
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.markDataPlaneCallClosed;
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.markExtProcStreamCompleted;
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.markExtProcStreamFailed;
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.outboundStreamToByteString;
import static io.grpc.xds.internal.extproc.ExternalProcessorUtil.toHeaderMap;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.ByteString;
import com.google.protobuf.Struct;
import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ProcessingMode;
import io.envoyproxy.envoy.service.ext_proc.v3.BodyMutation;
import io.envoyproxy.envoy.service.ext_proc.v3.BodyResponse;
import io.envoyproxy.envoy.service.ext_proc.v3.CommonResponse;
import io.envoyproxy.envoy.service.ext_proc.v3.ExternalProcessorGrpc;
import io.envoyproxy.envoy.service.ext_proc.v3.HttpBody;
import io.envoyproxy.envoy.service.ext_proc.v3.HttpHeaders;
import io.envoyproxy.envoy.service.ext_proc.v3.HttpTrailers;
import io.envoyproxy.envoy.service.ext_proc.v3.ImmediateResponse;
import io.envoyproxy.envoy.service.ext_proc.v3.ProcessingRequest;
import io.envoyproxy.envoy.service.ext_proc.v3.ProcessingResponse;
import io.envoyproxy.envoy.service.ext_proc.v3.ProtocolConfiguration;
import io.envoyproxy.envoy.service.ext_proc.v3.StreamedBodyResponse;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.Context;
import io.grpc.Deadline;
import io.grpc.DoubleHistogramMetricInstrument;
import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.MetricInstrumentRegistry;
import io.grpc.MetricRecorder;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.internal.DelayedClientCall;
import io.grpc.internal.SerializingExecutor;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import io.grpc.xds.ExternalProcessorFilter.ExternalProcessorFilterConfig;
import io.grpc.xds.Filter.FilterContext;
import io.grpc.xds.internal.extproc.DataPlaneCallState;
import io.grpc.xds.internal.extproc.EventType;
import io.grpc.xds.internal.extproc.ExtProcStreamState;
import io.grpc.xds.internal.extproc.KnownLengthInputStream;
import io.grpc.xds.internal.grpcservice.CachedChannelManager;
import io.grpc.xds.internal.grpcservice.HeaderValue;
import io.grpc.xds.internal.headermutations.HeaderMutationDisallowedException;
import io.grpc.xds.internal.headermutations.HeaderMutationFilter;
import io.grpc.xds.internal.headermutations.HeaderMutationRulesConfig;
import io.grpc.xds.internal.headermutations.HeaderMutator;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;

/**
 * Client-side interceptor for external processing filter.
 */
final class ExternalProcessorClientInterceptor implements ClientInterceptor {

  @VisibleForTesting
  static DoubleHistogramMetricInstrument clientHeadersDuration;
  @VisibleForTesting
  static DoubleHistogramMetricInstrument clientHalfCloseDuration;
  @VisibleForTesting
  static DoubleHistogramMetricInstrument serverHeadersDuration;
  @VisibleForTesting
  static DoubleHistogramMetricInstrument serverTrailersDuration;

  // Copied from io.grpc.opentelemetry.internal.OpenTelemetryConstants.LATENCY_BUCKETS
  private static final List<Double> LATENCY_BUCKETS = ImmutableList.of(
      0d,     0.00001d, 0.00005d, 0.0001d, 0.0003d, 0.0006d, 0.0008d, 0.001d, 0.002d,
      0.003d, 0.004d,   0.005d,   0.006d,  0.008d,  0.01d,   0.013d,  0.016d, 0.02d,
      0.025d, 0.03d,    0.04d,    0.05d,   0.065d,  0.08d,   0.1d,    0.13d,  0.16d,
      0.2d,   0.25d,    0.3d,     0.4d,    0.5d,    0.65d,   0.8d,    1d,     2d,
      5d,     10d,      20d,      50d,     100d);

  static {
    initMetricInstruments();
  }

  static synchronized void initMetricInstruments() {
    if (io.grpc.internal.GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)) {
      if (clientHeadersDuration == null) {
        MetricInstrumentRegistry registry = MetricInstrumentRegistry.getDefaultRegistry();

        clientHeadersDuration = registry.registerDoubleHistogram(
            "grpc.client_ext_proc.client_headers_duration",
            "Time between when the ext_proc filter sees the client's headers and when "
                + "it allows those headers to continue on to the next filter",
            "s",
            LATENCY_BUCKETS,
            ImmutableList.of("grpc.target"),
            ImmutableList.of("grpc.lb.backend_service"),
            true);

        clientHalfCloseDuration = registry.registerDoubleHistogram(
            "grpc.client_ext_proc.client_half_close_duration",
            "Time between when the ext_proc filter sees the client's half-close and when "
                + "it allows that half-close to continue on to the next filter",
            "s",
            LATENCY_BUCKETS,
            ImmutableList.of("grpc.target"),
            ImmutableList.of("grpc.lb.backend_service"),
            true);

        serverHeadersDuration = registry.registerDoubleHistogram(
            "grpc.client_ext_proc.server_headers_duration",
            "Time between when the ext_proc filter sees the server's headers and when "
                + "it allows those headers to continue on to the next filter",
            "s",
            LATENCY_BUCKETS,
            ImmutableList.of("grpc.target"),
            ImmutableList.of("grpc.lb.backend_service"),
            true);

        serverTrailersDuration = registry.registerDoubleHistogram(
            "grpc.client_ext_proc.server_trailers_duration",
            "Time between when the ext_proc filter sees the server's trailers and when "
                + "it allows those trailers to continue on to the next filter",
            "s",
            LATENCY_BUCKETS,
            ImmutableList.of("grpc.target"),
            ImmutableList.of("grpc.lb.backend_service"),
            true);
      }
    }
  }

  private final ExternalProcessorFilterConfig filterConfig;
  private final ScheduledExecutorService scheduler;
  private final MetricRecorder metricsRecorder;
  private final ManagedChannel extProcChannel;

  @VisibleForTesting
  ExternalProcessorClientInterceptor(ExternalProcessorFilterConfig filterConfig,
      CachedChannelManager cachedChannelManager,
      ScheduledExecutorService scheduler,
      FilterContext context) {
    this.filterConfig = filterConfig;
    checkNotNull(cachedChannelManager, "cachedChannelManager");
    this.scheduler = checkNotNull(scheduler, "scheduler");
    this.metricsRecorder = checkNotNull(context.metricsRecorder(), "metricsRecorder");
    this.extProcChannel = cachedChannelManager.getChannel(filterConfig.getGrpcServiceConfig());
  }

  @VisibleForTesting
  ExternalProcessorFilterConfig getFilterConfig() {
    return filterConfig;
  }

  @VisibleForTesting
  ManagedChannel getExtProcChannel() {
    return extProcChannel;
  }

  @Override
  @SuppressWarnings("unchecked")
  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
      MethodDescriptor<ReqT, RespT> method,
      CallOptions callOptions,
      Channel next) {
    SerializingExecutor serializingExecutor = new SerializingExecutor(callOptions.getExecutor());
    
    ExternalProcessorGrpc.ExternalProcessorStub extProcStub = ExternalProcessorGrpc.newStub(
        extProcChannel)
        .withExecutor(serializingExecutor);
    
    if (filterConfig.getGrpcServiceConfig().timeout().isPresent()) {
      long timeoutNanos = filterConfig.getGrpcServiceConfig().timeout().get().toNanos();
      if (timeoutNanos > 0) {
        extProcStub = extProcStub.withDeadlineAfter(timeoutNanos, TimeUnit.NANOSECONDS);
      }
    }
    if (filterConfig.getGrpcServiceConfig().initialMetadata() != null
        && !filterConfig.getGrpcServiceConfig().initialMetadata().isEmpty()) {
      Metadata extraHeaders = new Metadata();
      for (HeaderValue headerValue : filterConfig.getGrpcServiceConfig().initialMetadata()) {
        String key = headerValue.key();
        if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
          if (headerValue.rawValue().isPresent()) {
            Metadata.Key<byte[]> metadataKey =
                Metadata.Key.of(key, Metadata.BINARY_BYTE_MARSHALLER);
            extraHeaders.put(metadataKey, headerValue.rawValue().get().toByteArray());
          }
        } else {
          if (headerValue.value().isPresent()) {
            Metadata.Key<String> metadataKey =
                Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER);
            extraHeaders.put(metadataKey, headerValue.value().get());
          }
        }
      }
      extProcStub = extProcStub.withInterceptors(
          io.grpc.stub.MetadataUtils.newAttachHeadersInterceptor(extraHeaders));
    }


    // The filter chain is preceded by RawMessageClientInterceptor, so ReqT and RespT are
    // InputStream.
    MethodDescriptor<InputStream, InputStream> rawMethod =
        (MethodDescriptor<InputStream, InputStream>) (MethodDescriptor<?, ?>) method;
    ClientCall<InputStream, InputStream> rawCall =
        (ClientCall<InputStream, InputStream>) (ClientCall<?, ?>)
            next.newCall(method, callOptions);

    // Create a local subclass instance to buffer outbound actions
    DataPlaneDelayedCall<InputStream, InputStream> delayedCall =
        new DataPlaneDelayedCall<>(
            serializingExecutor, scheduler, callOptions.getDeadline());

    DataPlaneClientCall dataPlaneCall = new DataPlaneClientCall(
        delayedCall, rawCall, extProcStub, filterConfig, filterConfig.getMutationRulesConfig(),
        scheduler, rawMethod, next, metricsRecorder, next.authority(),
        callOptions.getOption(XdsNameResolver.CLUSTER_SELECTION_KEY));

    return (ClientCall<ReqT, RespT>) (ClientCall<?, ?>) dataPlaneCall;
  }

  // --- SHARED UTILITY METHODS ---

  /**
   * A local subclass to expose the protected constructor of DelayedClientCall.
   */
  private static class DataPlaneDelayedCall<ReqT, RespT> extends DelayedClientCall<ReqT, RespT> {
    DataPlaneDelayedCall(
        Executor executor, ScheduledExecutorService scheduler, @Nullable Deadline deadline) {
      super("ext_proc", executor, scheduler, deadline);
    }
  }

  /**
   * Handles the bidirectional stream with the External Processor.
   * Buffers the actual RPC start until the Ext Proc header response is received.
   */
  private static class DataPlaneClientCall 
      extends SimpleForwardingClientCall<InputStream, InputStream> {

    private final ExternalProcessorGrpc.ExternalProcessorStub stub;
    private final ExternalProcessorFilterConfig config;
    private final ClientCall<InputStream, InputStream> rawCall;
    private final DataPlaneDelayedCall<InputStream, InputStream> delayedCall;
    private final ScheduledExecutorService scheduler;
    private final Object streamLock = new Object();
    @Nullable private volatile EventType expectedRequestResponse;
    @Nullable private volatile EventType expectedResponseResponse;
    @Nullable private volatile ClientCallStreamObserver<ProcessingRequest>
        extProcClientCallRequestObserver;
    private final Queue<InputStream> pendingDrainingMessages =
        new ConcurrentLinkedQueue<>();
    @Nullable private volatile DataPlaneListener wrappedListener;
    private final HeaderMutationFilter mutationFilter;
    private final HeaderMutator mutator = HeaderMutator.create();
    private final AtomicInteger pendingRequests = new AtomicInteger(0);
    private final ProcessingMode currentProcessingMode;
    private final MethodDescriptor<?, ?> method;
    private final Channel channel;
    private final MetricRecorder metricsRecorder;
    private final String target;
    private final String backendService;
    private volatile Context callContext = Context.ROOT;

    private long clientHeadersStartNanos;
    private long clientHalfCloseStartNanos;
    private long serverHeadersStartNanos;
    private long serverTrailersStartNanos;

    private boolean protocolConfigSent = false;
    private ImmutableMap<String, Struct> collectedAttributes;
    private boolean requestAttributesSent = false;
    @Nullable private volatile Metadata requestHeaders;
    final AtomicReference<DataPlaneCallState> dataPlaneCallState =
        new AtomicReference<>(DataPlaneCallState.IDLE);
    final AtomicReference<ExtProcStreamState> extProcStreamState =
        new AtomicReference<>(ExtProcStreamState.ACTIVE);
    final AtomicBoolean passThroughMode = new AtomicBoolean(false);
    final AtomicBoolean requestSideClosed = new AtomicBoolean(false);
    final AtomicBoolean isProcessingTrailers = new AtomicBoolean(false);
    final AtomicBoolean pendingHalfClose = new AtomicBoolean(false);
    final AtomicBoolean bodyMessageSentToExtProc = new AtomicBoolean(false);
    private final AtomicBoolean downstreamCancelled = new AtomicBoolean(false);

    protected DataPlaneClientCall(
        DataPlaneDelayedCall<InputStream, InputStream> delayedCall,
        ClientCall<InputStream, InputStream> rawCall,
        ExternalProcessorGrpc.ExternalProcessorStub stub,
        ExternalProcessorFilterConfig config,
        Optional<HeaderMutationRulesConfig> mutationRulesConfig,
        ScheduledExecutorService scheduler,
        MethodDescriptor<?, ?> method,
        Channel channel,
        MetricRecorder metricsRecorder,
        String target,
        String backendService) {
      super(delayedCall);
      this.delayedCall = delayedCall;
      this.rawCall = rawCall;
      this.stub = stub;
      this.config = config;
      this.currentProcessingMode = config.getExternalProcessor().getProcessingMode();
      this.mutationFilter = new HeaderMutationFilter(mutationRulesConfig);
      this.scheduler = scheduler;
      this.method = method;
      this.channel = channel;
      this.metricsRecorder = checkNotNull(metricsRecorder, "metricsRecorder");
      this.target = checkNotNull(target, "target");
      this.backendService = checkNotNull(backendService, "backendService");
    }



    private void activateCall() {
      if ((extProcStreamState.get() == ExtProcStreamState.FAILED
              && !config.getFailureModeAllow()
              && !config.getObservabilityMode())
          || !dataPlaneCallState.compareAndSet(
              DataPlaneCallState.IDLE, DataPlaneCallState.ACTIVE)) {
        return;
      }
      if (clientHeadersStartNanos > 0) {
        long durationNanos = System.nanoTime() - clientHeadersStartNanos;
        recordDuration(clientHeadersDuration, durationNanos);
        clientHeadersStartNanos = 0;
      }
      Runnable toRun = delayedCall.setCall(rawCall);
      if (toRun != null) {
        callContext.run(toRun);
      }
      drainPendingRequests();
      onReadyNotify();
    }

    private void recordDuration(DoubleHistogramMetricInstrument instrument, long durationNanos) {
      if (instrument != null) {
        double durationSecs = (double) durationNanos / 1_000_000_000.0;
        metricsRecorder.recordDoubleHistogram(
            instrument,
            durationSecs,
            ImmutableList.of(target),
            ImmutableList.of(backendService));
      }
    }

    /**
     * Validates whether the body response uses unsupported gRPC message compression.
     * If compression is unsupported, this method will cancel the call, transition the
     * stream to a failed state, send an error to the external processor, and return false.
     *
     * @param bodyResponse the response to validate
     * @return true if validation passes (compression is supported or not used),
     *     false if validation fails
     */
    private boolean validateCompressionSupport(BodyResponse bodyResponse) {
      if (bodyResponse.hasResponse() && bodyResponse.getResponse().hasBodyMutation()) {
        BodyMutation mutation = 
            bodyResponse.getResponse().getBodyMutation();
        if (mutation.hasStreamedResponse()
            && mutation.getStreamedResponse().getGrpcMessageCompressed()) {
          StatusRuntimeException ex = Status.UNAVAILABLE
              .withDescription("gRPC message compression not supported in ext_proc")
              .asRuntimeException();
          synchronized (streamLock) {
            if (!extProcStreamState.get().isCompleted()
                && extProcClientCallRequestObserver != null) {
              extProcClientCallRequestObserver.onError(ex);
            }
          }
          activateCall();
          markExtProcStreamFailed(extProcStreamState);
          cancelDownstream("gRPC message compression not supported in ext_proc", ex);
          closeExtProcStream();
          return false;
        }
      }
      return true;
    }



    @Override
    public void start(Listener<InputStream> responseListener, Metadata headers) {
      this.callContext = Context.current();
      clientHeadersStartNanos = System.nanoTime();
      this.requestHeaders = headers;
      this.wrappedListener = new DataPlaneListener(responseListener, rawCall, this);

      // DelayedClientCall.start will buffer the listener and headers until setCall is called.
      super.start(wrappedListener, headers);

      stub.process(new ClientResponseObserver<ProcessingRequest, ProcessingResponse>() {
        @Override
        public void beforeStart(ClientCallStreamObserver<ProcessingRequest> requestStream) {
          synchronized (streamLock) {
            extProcClientCallRequestObserver = requestStream;
          }
          requestStream.setOnReadyHandler(DataPlaneClientCall.this::onExtProcStreamReady);
        }

        @Override
        public void onNext(ProcessingResponse response) {
          try {
            if (config.getObservabilityMode()) {
              return;
            }

            if (response.hasImmediateResponse()) {
              if (config.getDisableImmediateResponse()) {
                internalOnError(Status.UNAVAILABLE
                    .withDescription(
                        "Immediate response is disabled but received from external processor")
                    .asRuntimeException());
                return;
              }
              handleImmediateResponse(response.getImmediateResponse(), wrappedListener);
              return;
            }

            if (response.hasRequestHeaders()) {
              EventType expected = expectedRequestResponse;
              if (expected == null || expected != EventType.REQUEST_HEADERS) {
                internalOnError(Status.UNAVAILABLE
                    .withDescription("Protocol error: received response out of order. Expected: " 
                        + expected + ", Received: REQUEST_HEADERS")
                    .asRuntimeException());
                return;
              }
              expectedRequestResponse = null;
            } else if (response.hasResponseHeaders()) {
              EventType expected = expectedResponseResponse;
              if (expected == null || expected != EventType.RESPONSE_HEADERS) {
                internalOnError(Status.UNAVAILABLE
                    .withDescription("Protocol error: received response out of order. Expected: " 
                        + expected + ", Received: RESPONSE_HEADERS")
                    .asRuntimeException());
                return;
              }
              expectedResponseResponse = null;
            } else if (response.hasResponseTrailers()) {
              EventType expected = expectedResponseResponse;
              if (expected == null || expected != EventType.RESPONSE_TRAILERS) {
                internalOnError(Status.UNAVAILABLE
                    .withDescription("Protocol error: received response out of order. Expected: " 
                        + expected + ", Received: RESPONSE_TRAILERS")
                    .asRuntimeException());
                return;
              }
              expectedResponseResponse = null;
            } else if (response.hasRequestBody()) {
              EventType expected = expectedRequestResponse;
              if (expected == EventType.REQUEST_HEADERS) {
                internalOnError(Status.UNAVAILABLE
                    .withDescription(
                        "Protocol error: received request_body before request_headers response.")
                    .asRuntimeException());
                return;
              }
            } else if (response.hasResponseBody()) {
              EventType expected = expectedResponseResponse;
              if (expected == EventType.RESPONSE_HEADERS) {
                internalOnError(Status.UNAVAILABLE
                    .withDescription(
                        "Protocol error: received response_body before headers response.")
                    .asRuntimeException());
                return;
              }
            }

            if (response.getRequestDrain()) {
              extProcStreamState.set(ExtProcStreamState.DRAINING);
              halfCloseExtProcStream();
              activateCall();
            }

            // 1. Client Headers
            if (response.hasRequestHeaders()) {
              if (response.getRequestHeaders().hasResponse()) {
                if (response.getRequestHeaders().getResponse().getStatus()
                    == CommonResponse.ResponseStatus.CONTINUE_AND_REPLACE) {
                  internalOnError(Status.UNAVAILABLE
                      .withDescription("CONTINUE_AND_REPLACE is not supported")
                      .asRuntimeException());
                  return;
                }
                applyHeaderMutations(
                    requestHeaders,
                    response.getRequestHeaders().getResponse().getHeaderMutation(),
                    mutationFilter,
                    mutator);
              }
              activateCall();
            }
            // 2. Client Message (Request Body)
            else if (response.hasRequestBody()) {
              if (validateCompressionSupport(response.getRequestBody())) {
                handleRequestBodyResponse(response.getRequestBody());
              }
            }
            // 4. Server Headers
            else if (response.hasResponseHeaders()) {
              if (response.getResponseHeaders().hasResponse()) {
                if (response.getResponseHeaders().getResponse().getStatus()
                    == CommonResponse.ResponseStatus.CONTINUE_AND_REPLACE) {
                  internalOnError(Status.UNAVAILABLE
                      .withDescription("CONTINUE_AND_REPLACE is not supported")
                      .asRuntimeException());
                  return;
                }
                Metadata target = wrappedListener.isTrailersOnly()
                    ? wrappedListener.getSavedTrailers() : wrappedListener.getSavedHeaders();
                applyHeaderMutations(
                    target,
                    response.getResponseHeaders().getResponse().getHeaderMutation(),
                    mutationFilter,
                    mutator);
              }
              if (wrappedListener.isTrailersOnly()) {
                wrappedListener.proceedWithClose();
              } else {
                wrappedListener.proceedWithHeaders();
              }
            }
            // 5. Server Message (Response Body)
            else if (response.hasResponseBody()) {
              if (validateCompressionSupport(response.getResponseBody())) {
                handleResponseBodyResponse(response.getResponseBody(), wrappedListener);
              }
            }
            // 6. Response Trailers
            else if (response.hasResponseTrailers()) {
              if (response.getResponseTrailers().hasHeaderMutation()) {
                applyHeaderMutations(
                    wrappedListener.getSavedTrailers(),
                    response.getResponseTrailers().getHeaderMutation(),
                    mutationFilter,
                    mutator);
              }
              wrappedListener.proceedWithClose();
            }

            checkEndOfStream(response);
          } catch (Throwable t) {
            internalOnError(t);
          }
        }

        @Override
        public void onError(Throwable t) {
          if (markExtProcStreamFailed(extProcStreamState)) {
            synchronized (streamLock) {
              extProcClientCallRequestObserver = null;
            }
            if (config.getObservabilityMode()
                || (config.getFailureModeAllow() && !bodyMessageSentToExtProc.get())) {
              handleFailOpen(wrappedListener);
            } else {
              String message = "External processor stream failed";
              cancelDownstream(message, t);
              wrappedListener.proceedWithClose();
            }
          }
        }

        @Override
        public void onCompleted() {
          if (markExtProcStreamCompleted(extProcStreamState)) {
            handleFailOpen(wrappedListener);
          }
        }
      });

      this.collectedAttributes = collectAttributes(
          config.getRequestAttributes(), method, channel.authority(), headers);

      boolean sendRequestHeaders =
          currentProcessingMode.getRequestHeaderMode() == ProcessingMode.HeaderSendMode.SEND
          || currentProcessingMode.getRequestHeaderMode()
              == ProcessingMode.HeaderSendMode.DEFAULT;

      if (sendRequestHeaders) {
        sendToExtProc(ProcessingRequest.newBuilder()
            .setRequestHeaders(HttpHeaders.newBuilder()
                .setHeaders(toHeaderMap(headers, config.getForwardRulesConfig()))
                .setEndOfStream(false)
                .build())
            .build());
      }

      if (config.getObservabilityMode() || !sendRequestHeaders) {
        activateCall();
      }
    }

    private void sendToExtProc(ProcessingRequest request) {
      synchronized (streamLock) {
        if (extProcStreamState.get().isCompleted()) {
          return;
        }
        
        if (request.hasRequestHeaders()) {
          expectedRequestResponse = EventType.REQUEST_HEADERS;
        } else if (request.hasResponseHeaders()) {
          expectedResponseResponse = EventType.RESPONSE_HEADERS;
        } else if (request.hasResponseTrailers()) {
          expectedResponseResponse = EventType.RESPONSE_TRAILERS;
        }

        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(requestToSend)
              .setObservabilityMode(true)
              .build();
        }

        extProcClientCallRequestObserver.onNext(requestToSend);
      }
    }

    private void onExtProcStreamReady() {
      drainPendingRequests();
      onReadyNotify();
    }

    private void drainPendingRequests() {
      int toRequest = pendingRequests.getAndSet(0);
      if (toRequest > 0) {
        super.request(toRequest);
      }
    }

    private void closeExtProcStream() {
      synchronized (streamLock) {
        if (markExtProcStreamCompleted(extProcStreamState)) {
          if (extProcClientCallRequestObserver != null) {
            extProcClientCallRequestObserver.onCompleted();
          }
        }
      }
    }

    private void internalOnError(Throwable t) {
      if (markExtProcStreamFailed(extProcStreamState)) {
        synchronized (streamLock) {
          if (extProcClientCallRequestObserver != null) {
            try {
              extProcClientCallRequestObserver.onError(t);
            } catch (Throwable ignored) {
              // Ignore exceptions during cancel/onError propagation
            }
            extProcClientCallRequestObserver = null;
          }
        }
        if (config.getObservabilityMode()
            || (config.getFailureModeAllow() && !bodyMessageSentToExtProc.get())) {
          handleFailOpen(wrappedListener);
        } else {
          String message = "External processor stream failed";
          cancelDownstream(message, t);
          wrappedListener.proceedWithClose();
        }
      }
    }

    private void halfCloseExtProcStream() {
      synchronized (streamLock) {
        if (!extProcStreamState.get().isCompleted() && extProcClientCallRequestObserver != null) {
          extProcClientCallRequestObserver.onCompleted();
        }
      }
    }

    private void onReadyNotify() {
      wrappedListener.onReadyNotify();
    }

    private boolean isSidecarReady() {
      ExtProcStreamState state = extProcStreamState.get();
      if (state.isCompleted()) {
        return true;
      }
      if (state.isDraining()) {
        return false;
      }
      synchronized (streamLock) {
        ClientCallStreamObserver<ProcessingRequest> observer = extProcClientCallRequestObserver;
        return observer != null && observer.isReady();
      }
    }

    @Override
    public boolean isReady() {
      if (passThroughMode.get()) {
        return super.isReady();
      }
      if (extProcStreamState.get().isCompleted()) {
        return super.isReady();
      }
      if (dataPlaneCallState.get() == DataPlaneCallState.IDLE && !config.getObservabilityMode()) {
        return false;
      }
      boolean sidecarReady = isSidecarReady();
      if (config.getObservabilityMode()) {
        return super.isReady() && sidecarReady;
      }
      return sidecarReady;
    }

    @Override
    public void request(int numMessages) {
      if (passThroughMode.get() || extProcStreamState.get().isCompleted()) {
        super.request(numMessages);
        return;
      }
      if (!config.getObservabilityMode() 
          && currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.GRPC) {
        super.request(numMessages);
        return;
      }
      if (!isSidecarReady()) {
        pendingRequests.addAndGet(numMessages);
        return;
      }
      super.request(numMessages);
    }

    @Override
    public void sendMessage(InputStream message) {
      if (requestSideClosed.get()) {
        // External processor already closed the request stream. Discard further messages.
        return;
      }

      if (passThroughMode.get()) {
        super.sendMessage(message);
        return;
      }

      synchronized (streamLock) {
        if (passThroughMode.get()) {
          super.sendMessage(message);
          return;
        }

        ExtProcStreamState state = extProcStreamState.get();
        if (state.isDraining() || state.isCompleted()) {
          if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) {
            super.sendMessage(message);
            return;
          }
          try {
            ByteString copiedBody = ByteString.readFrom(message);
            pendingDrainingMessages.add(new KnownLengthInputStream(copiedBody));
          } catch (IOException e) {
            rawCall.cancel("Failed to copy outbound message for buffering", e);
          }
          return;
        }
      }

      if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) {
        super.sendMessage(message);
        return;
      }

      // Mode is GRPC
      try {
        ByteString bodyByteString = outboundStreamToByteString(message);
        sendToExtProc(ProcessingRequest.newBuilder()
            .setRequestBody(HttpBody.newBuilder()
                .setBody(bodyByteString)
                .setEndOfStream(false)
                .build())
            .build());
        bodyMessageSentToExtProc.set(true);

        if (config.getObservabilityMode()) {
          super.sendMessage(new KnownLengthInputStream(bodyByteString));
        }
      } catch (IOException e) {
        rawCall.cancel("Failed to serialize message for External Processor", e);
      }
    }

    private void proceedWithHalfClose() {
      if (clientHalfCloseStartNanos > 0) {
        long durationNanos = System.nanoTime() - clientHalfCloseStartNanos;
        recordDuration(clientHalfCloseDuration, durationNanos);
        clientHalfCloseStartNanos = 0;
      }
      super.halfClose();
    }

    @Override
    public void halfClose() {
      clientHalfCloseStartNanos = System.nanoTime();
      if (passThroughMode.get()) {
        if (requestSideClosed.compareAndSet(false, true)) {
          proceedWithHalfClose();
        }
        return;
      }

      pendingHalfClose.set(true);

      if (extProcStreamState.get().isCompleted()) {
        if (passThroughMode.get()) {
          if (requestSideClosed.compareAndSet(false, true)) {
            proceedWithHalfClose();
          }
        }
        return;
      }

      if (extProcStreamState.get().isDraining()) {
        boolean canProceed = false;
        synchronized (streamLock) {
          if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE
              || (!bodyMessageSentToExtProc.get() && pendingDrainingMessages.isEmpty())) {
            canProceed = true;
          }
        }
        if (canProceed) {
          if (requestSideClosed.compareAndSet(false, true)) {
            proceedWithHalfClose();
          }
        }
        return;
      }

      if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) {
        if (requestSideClosed.compareAndSet(false, true)) {
          proceedWithHalfClose();
        }
        return;
      }

      // Mode is GRPC
      sendToExtProc(ProcessingRequest.newBuilder()
          .setRequestBody(HttpBody.newBuilder()
              .setEndOfStreamWithoutMessage(true)
              .build())
          .build());
    }

    private void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
      if (downstreamCancelled.compareAndSet(false, true)) {
        delayedCall.cancel(message, cause);
      }
    }

    @Override
    public void cancel(@Nullable String message, @Nullable Throwable cause) {
      synchronized (streamLock) {
        if (!extProcStreamState.get().isCompleted() && extProcClientCallRequestObserver != null) {
          extProcClientCallRequestObserver.onError(
              Status.CANCELLED
                  .withDescription(message)
                  .withCause(cause)
                  .asRuntimeException());
        }
      }
      cancelDownstream(message, cause);
    }

    private void handleRequestBodyResponse(BodyResponse bodyResponse) {
      if (bodyResponse.hasResponse() && bodyResponse.getResponse().hasBodyMutation()) {
        BodyMutation mutation = bodyResponse.getResponse().getBodyMutation();
        if (mutation.hasStreamedResponse()) {
          StreamedBodyResponse streamed = mutation.getStreamedResponse();
          if (!streamed.getEndOfStreamWithoutMessage()) {
            super.sendMessage(new KnownLengthInputStream(streamed.getBody()));
          }
          if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) {
            if (requestSideClosed.compareAndSet(false, true)) {
              proceedWithHalfClose();
            }
          }
        }
      }
    }

    private void handleResponseBodyResponse(
        BodyResponse bodyResponse, DataPlaneListener listener) {
      if (bodyResponse.hasResponse() && bodyResponse.getResponse().hasBodyMutation()) {
        BodyMutation mutation = bodyResponse.getResponse().getBodyMutation();
        if (mutation.hasStreamedResponse()) {
          StreamedBodyResponse streamed = mutation.getStreamedResponse();
          listener.onExternalBody(streamed.getBody());
        }
      }
    }

    private void handleImmediateResponse(ImmediateResponse immediate, DataPlaneListener listener)
        throws HeaderMutationDisallowedException {
      Status status = Status.fromCodeValue(immediate.getGrpcStatus().getStatus());
      if (!immediate.getDetails().isEmpty()) {
        status = status.withDescription(immediate.getDetails());
      }

      Metadata trailers = new Metadata();
      if (immediate.hasHeaders()) {
        applyHeaderMutations(trailers, immediate.getHeaders(), mutationFilter, mutator);
      }

      listener.setImmediateResponse(status, trailers);

      if (isProcessingTrailers.get()) {
        // If sent in response to a server trailers event, sets the status and optionally
        // headers to be included in the trailers.
        listener.unblockAfterStreamComplete();
      } else {
        // If sent in response to any other event, it will cause the data plane RPC to
        // immediately fail with the specified status as if it were an out-of-band
        // cancellation.
        rawCall.cancel(status.getDescription(), null);
        listener.unblockAfterStreamComplete();
      }
      closeExtProcStream();
    }

    private void drainPendingDrainingMessages() {
      synchronized (streamLock) {
        InputStream msg;
        while ((msg = pendingDrainingMessages.poll()) != null) {
          super.sendMessage(msg);
        }
        passThroughMode.set(true);
        if (pendingHalfClose.get()) {
          if (requestSideClosed.compareAndSet(false, true)) {
            proceedWithHalfClose();
          }
        }
      }
    }

    private void handleFailOpen(DataPlaneListener listener) {
      activateCall();
      drainPendingRequests();
      listener.unblockAfterStreamComplete();
      closeExtProcStream();
    }

    private void checkEndOfStream(ProcessingResponse response) {
      boolean terminal = false;
      if (response.hasResponseTrailers()) {
        terminal = true;
      } else if (response.hasResponseHeaders() && wrappedListener.isTrailersOnly()) {
        terminal = true;
      }

      if (terminal) {
        wrappedListener.unblockAfterStreamComplete();
        closeExtProcStream();
      }
    }

    long getServerHeadersStartNanos() {
      return serverHeadersStartNanos;
    }

    void setServerHeadersStartNanos(long serverHeadersStartNanos) {
      this.serverHeadersStartNanos = serverHeadersStartNanos;
    }

    long getServerTrailersStartNanos() {
      return serverTrailersStartNanos;
    }

    void setServerTrailersStartNanos(long serverTrailersStartNanos) {
      this.serverTrailersStartNanos = serverTrailersStartNanos;
    }

    AtomicReference<ExtProcStreamState> getExtProcStreamState() {
      return extProcStreamState;
    }

    ProcessingMode getCurrentProcessingMode() {
      return currentProcessingMode;
    }

    AtomicBoolean getPassThroughMode() {
      return passThroughMode;
    }

    ExternalProcessorFilterConfig getConfig() {
      return config;
    }

    Context getCallContext() {
      return callContext;
    }

    ScheduledExecutorService getScheduler() {
      return scheduler;
    }

    AtomicBoolean getIsProcessingTrailers() {
      return isProcessingTrailers;
    }
  }

  private static class DataPlaneListener extends SimpleForwardingClientCallListener<InputStream> {
    private final ClientCall<?, ?> rawCall;
    private final DataPlaneClientCall dataPlaneClientCall;
    private final Queue<InputStream> savedMessages = new ConcurrentLinkedQueue<>();
    private boolean inboundPassThrough = false;
    @Nullable private volatile Metadata savedHeaders;
    @Nullable private volatile Metadata savedTrailers;
    @Nullable private volatile Status savedStatus;
    private final AtomicBoolean terminationTriggered = new AtomicBoolean(false);
    private final AtomicBoolean responseHeadersSent = new AtomicBoolean(false);
    private final AtomicBoolean trailersOnly = new AtomicBoolean(false);

    protected DataPlaneListener(
        ClientCall.Listener<InputStream> delegate,
        ClientCall<?, ?> rawCall,
        DataPlaneClientCall dataPlaneClientCall) {
      super(delegate);
      this.rawCall = rawCall;
      this.dataPlaneClientCall = dataPlaneClientCall;
    }

    boolean isTrailersOnly() {
      return trailersOnly.get();
    }

    Metadata getSavedHeaders() {
      return savedHeaders;
    }

    Metadata getSavedTrailers() {
      return savedTrailers;
    }

    void setImmediateResponse(Status status, Metadata trailers) {
      this.savedStatus = status;
      this.savedTrailers = trailers;
    }

    @Override
    public void onReady() {
      dataPlaneClientCall.drainPendingRequests();
      onReadyNotify();
    }

    @Override
    public void onHeaders(Metadata headers) {
      dataPlaneClientCall.setServerHeadersStartNanos(System.nanoTime());
      responseHeadersSent.set(true);
      boolean sendResponseHeaders =
          dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode()
              == ProcessingMode.HeaderSendMode.SEND
          || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode()
              == ProcessingMode.HeaderSendMode.DEFAULT;

      if (dataPlaneClientCall.getExtProcStreamState().get().isDraining() && sendResponseHeaders) {
        this.savedHeaders = headers;
        return;
      }

      if (dataPlaneClientCall.getPassThroughMode().get() 
          || dataPlaneClientCall.getExtProcStreamState().get().isCompleted() 
          || !sendResponseHeaders) {
        proceedWithHeaders(headers);
        return;
      }

      this.savedHeaders = headers;
      dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder()
          .setResponseHeaders(HttpHeaders.newBuilder()
              .setHeaders(
                  toHeaderMap(headers, dataPlaneClientCall.getConfig().getForwardRulesConfig()))
              .build())
          .build());

      if (dataPlaneClientCall.getConfig().getObservabilityMode()) {
        proceedWithHeaders();
      }
    }

    @Override
    public void onMessage(InputStream message) {
      synchronized (savedMessages) {
        if (inboundPassThrough) {
          dataPlaneClientCall.getCallContext().run(() -> delegate().onMessage(message));
          return;
        }

        boolean checkDrain = dataPlaneClientCall.getExtProcStreamState().get().isDraining()
            && dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode()
                == ProcessingMode.BodySendMode.GRPC;

        if (savedHeaders != null || checkDrain) {
          try {
            ByteString copiedBody = ByteString.readFrom(message);
            savedMessages.add(new KnownLengthInputStream(copiedBody));
          } catch (IOException e) {
            rawCall.cancel("Failed to copy inbound message for buffering", e);
          }
          return;
        }
      }

      if (dataPlaneClientCall.getPassThroughMode().get()) {
        dataPlaneClientCall.getCallContext().run(() -> delegate().onMessage(message));
        return;
      }

      if (dataPlaneClientCall.getExtProcStreamState().get().isCompleted()
          || dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode()
              != ProcessingMode.BodySendMode.GRPC) {
        dataPlaneClientCall.getCallContext().run(() -> delegate().onMessage(message));
        return;
      }

      try {
        ByteString bodyByteString = ByteString.readFrom(message);
        sendResponseBodyToExtProc(bodyByteString, false);
        dataPlaneClientCall.bodyMessageSentToExtProc.set(true);

        if (dataPlaneClientCall.getConfig().getObservabilityMode()) {
          // If needed, downstream reading can be made more optimal by creating a wrapped
          // Inputstream wraps the underlying bytestring and that implements HasByteBuffer,
          // Detachable, KnownLength
          dataPlaneClientCall.getCallContext().run(
              () -> delegate().onMessage(bodyByteString.newInput()));
        }
      } catch (IOException e) {
        rawCall.cancel("Failed to read server response", e);
      }
    }

    @Override
    public void onClose(Status status, Metadata trailers) {
      dataPlaneClientCall.setServerTrailersStartNanos(System.nanoTime());
      ExtProcStreamState extProcStreamState =
          dataPlaneClientCall.getExtProcStreamState().get();
      if (extProcStreamState.isFailed()
          && !dataPlaneClientCall.getConfig().getObservabilityMode()
          && (!dataPlaneClientCall.getConfig().getFailureModeAllow()
              || dataPlaneClientCall.bodyMessageSentToExtProc.get())) {
        if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) {
          proceedWithClose(Status.INTERNAL.withDescription("External processor stream failed")
              .withCause(status.getCause()), new Metadata());
        }
        return;
      }
      if (dataPlaneClientCall.getPassThroughMode().get()) {
        if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) {
          proceedWithClose(status, trailers);
        }
        return;
      }

      if (this.savedStatus == null) {
        this.savedStatus = status;
        this.savedTrailers = trailers;
      }

      // If we are still waiting for the external processor to validate response headers,
      // buffer the close status/trailers and defer the close until headers are processed.
      if (savedHeaders != null) {
        return;
      }

      boolean sendResponseTrailers =
          dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode()
              == ProcessingMode.HeaderSendMode.SEND;

      if (dataPlaneClientCall.getExtProcStreamState().get().isDraining() && sendResponseTrailers) {
        return;
      }

      if (!responseHeadersSent.get()) {
        trailersOnly.set(true);
      }

      triggerCloseHandshake();
    }

    void onReadyNotify() {
      dataPlaneClientCall.getCallContext().run(() -> delegate().onReady());
    }

    void proceedWithHeaders() {
      if (savedHeaders != null) {
        proceedWithHeaders(savedHeaders);
        synchronized (savedMessages) {
          savedHeaders = null;
          if (!dataPlaneClientCall.getExtProcStreamState().get().isDraining()) {
            InputStream msg;
            while ((msg = savedMessages.poll()) != null) {
              onMessage(msg);
            }
          }
        }
        onReadyNotify();
        if (savedStatus != null) {
          triggerCloseHandshake();
        }
      }
    }

    private void proceedWithHeaders(Metadata headers) {
      if (dataPlaneClientCall.getServerHeadersStartNanos() > 0) {
        long durationNanos = System.nanoTime() - dataPlaneClientCall.getServerHeadersStartNanos();
        dataPlaneClientCall.recordDuration(serverHeadersDuration, durationNanos);
        dataPlaneClientCall.setServerHeadersStartNanos(0);
      }
      dataPlaneClientCall.getCallContext().run(() -> delegate().onHeaders(headers));
    }

    void proceedWithClose() {
      if (savedStatus != null) {
        if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) {
          proceedWithClose(savedStatus, savedTrailers);
        }
        savedStatus = null;
        savedTrailers = null;
      }
    }

    private void proceedWithClose(Status status, Metadata trailers) {
      if (dataPlaneClientCall.getServerTrailersStartNanos() > 0) {
        long durationNanos = System.nanoTime() - dataPlaneClientCall.getServerTrailersStartNanos();
        dataPlaneClientCall.recordDuration(serverTrailersDuration, durationNanos);
        dataPlaneClientCall.setServerTrailersStartNanos(0);
      }
      dataPlaneClientCall.getCallContext().run(() -> delegate().onClose(status, trailers));
    }

    void onExternalBody(ByteString body) {
      // If needed, downstream reading can be made more optimal by creating a wrapped
      // Inputstream wraps the underlying bytestring and that implements HasByteBuffer,
      // Detachable, KnownLength
      dataPlaneClientCall.getCallContext().run(
          () -> delegate().onMessage(body.newInput()));
    }

    void unblockAfterStreamComplete() {
      proceedWithHeaders();
      proceedWithSavedMessages();
      dataPlaneClientCall.drainPendingDrainingMessages();
      proceedWithClose();
    }

    private void proceedWithSavedMessages() {
      synchronized (savedMessages) {
        InputStream msg;
        while ((msg = savedMessages.poll()) != null) {
          final InputStream finalMsg = msg;
          dataPlaneClientCall.getCallContext().run(() -> delegate().onMessage(finalMsg));
        }
        inboundPassThrough = true;
      }
    }

    private void triggerCloseHandshake() {
      if (dataPlaneClientCall.getExtProcStreamState().get().isCompleted()
          || !terminationTriggered.compareAndSet(false, true)) {
        return;
      }

      boolean sendResponseHeaders =
          dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode()
              == ProcessingMode.HeaderSendMode.SEND
          || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode()
              == ProcessingMode.HeaderSendMode.DEFAULT;

      boolean sendResponseTrailers =
          dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode()
              == ProcessingMode.HeaderSendMode.SEND;

      if (trailersOnly.get()) {
        if (sendResponseHeaders) {
          dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder()
              .setResponseHeaders(HttpHeaders.newBuilder()
                  .setHeaders(
                      toHeaderMap(
                          savedTrailers,
                          dataPlaneClientCall.getConfig().getForwardRulesConfig()))
                  .setEndOfStream(true)
                  .build())
              .build());
        } else {
          proceedWithClose();
          if (!dataPlaneClientCall.getConfig().getObservabilityMode()) {
            dataPlaneClientCall.closeExtProcStream();
          }
        }
      } else if (sendResponseTrailers) {
        dataPlaneClientCall.getIsProcessingTrailers().set(true);
        dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder()
            .setResponseTrailers(HttpTrailers.newBuilder()
                .setTrailers(
                    toHeaderMap(
                        savedTrailers,
                        dataPlaneClientCall.getConfig().getForwardRulesConfig()))
                .build())
            .build());
      } else {
        proceedWithClose();
        if (!dataPlaneClientCall.getConfig().getObservabilityMode()) {
          dataPlaneClientCall.closeExtProcStream();
        }
      }

      if (dataPlaneClientCall.getConfig().getObservabilityMode()) {
        proceedWithClose();
        @SuppressWarnings("unused")
        ScheduledFuture<?> unused = dataPlaneClientCall.getScheduler().schedule(
            dataPlaneClientCall::closeExtProcStream,
            dataPlaneClientCall.getConfig().getDeferredCloseTimeoutNanos(),
            TimeUnit.NANOSECONDS);
      }
    }

    private void sendResponseBodyToExtProc(
        @Nullable ByteString bodyByteString, boolean endOfStream) {
      if (dataPlaneClientCall.getExtProcStreamState().get().isCompleted()
          || dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode()
              != ProcessingMode.BodySendMode.GRPC) {
        return;
      }

      HttpBody.Builder bodyBuilder =
          HttpBody.newBuilder();
      if (bodyByteString != null) {
        bodyBuilder.setBody(bodyByteString);
      }
      bodyBuilder.setEndOfStream(endOfStream);

      dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder()
          .setResponseBody(bodyBuilder.build())
          .build());
    }
  }
}

grpc / grpc-java / #20413

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous