fix: retry OTLP gRPC responses with trailer status

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## Unreleased
 
+### Exporters
+
+* OTLP gRPC: Retry responses that report a retryable status in trailers
+  ([#8854](https://github.com/open-telemetry/opentelemetry-java/pull/8854)).
+
 ## Version 1.66.0 (2026-09-11)
 
 ### API

exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/OkHttpGrpcSender.java

@@ -57,7 +57,9 @@
 import okhttp3.Callback;
 import okhttp3.ConnectionSpec;
 import okhttp3.Dispatcher;
+import okhttp3.Headers;
 import okhttp3.HttpUrl;
+import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
 import okhttp3.Protocol;
 import okhttp3.Request;
@@ -66,6 +68,7 @@
 import okhttp3.ResponseBody;
 import okhttp3.TlsVersion;
 import okio.Buffer;
+import okio.BufferedSource;
 import okio.GzipSource;
 
 /**
@@ -122,7 +125,10 @@ public OkHttpGrpcSender(
     if (retryPolicy != null) {
       clientBuilder.addInterceptor(
           new RetryInterceptor(
-              retryPolicy, OkHttpGrpcSender::isRetryable, response -> OptionalLong.empty()));
+              retryPolicy,
+              OkHttpGrpcSender::isRetryable,
+              response -> OptionalLong.empty(),
+              response -> prepareResponseForRetry(response, maxResponseBodySize)));
     }
 
     boolean isPlainHttp = endpoint.startsWith("http://");
@@ -366,13 +372,66 @@ public CompletableResultCode shutdown() {
 
   /** Whether response is retriable or not. */
   public static boolean isRetryable(Response response) {
-    // We don't check trailers for retry since retryable error codes always come with response
-    // headers, not trailers, in practice.
     String grpcStatus = response.header(GRPC_STATUS);
     if (grpcStatus == null) {
-      return false;
+      try {
+        grpcStatus = response.trailers().get(GRPC_STATUS);
+      } catch (IOException | IllegalStateException e) {
+        return false;
+      }
+    }
+    return grpcStatus != null && RetryUtil.retryableGrpcStatusCodes().contains(grpcStatus);
+  }
+
+  private static Response prepareResponseForRetry(Response response, long maxResponseBodySize)
+      throws IOException {
+    if (response.header(GRPC_STATUS) != null) {
+      return response;
+    }
+
+    ResponseBody body = response.body();
+    Buffer buffer = new Buffer();
+    long readUpTo =
+        maxResponseBodySize >= Long.MAX_VALUE - 5 ? Long.MAX_VALUE : maxResponseBodySize + 6;
+    while (buffer.size() < readUpTo) {
+      long read = body.source().read(buffer, readUpTo - buffer.size());
+      if (read == -1L) {
+        break;
+      }
+    }
+
+    boolean responseBodyTooLarge = buffer.size() > maxResponseBodySize;
+    Headers trailers = responseBodyTooLarge ? Headers.of() : response.trailers();
+    Buffer replacementBuffer = buffer;
+    ResponseBody replacementBody =
+        new ResponseBody() {
+          @Override
+          public long contentLength() {
+            return replacementBuffer.size();
+          }
+
+          @Override
+          public MediaType contentType() {
+            return body.contentType();
+          }
+
+          @Override
+          public BufferedSource source() {
+            return replacementBuffer;
+          }
+        };
+    Response.Builder responseBuilder = response.newBuilder();
+    response.close();
+    responseBuilder.body(replacementBody);
+    String grpcStatus = trailers.get(GRPC_STATUS);
+    if (grpcStatus != null) {
+      responseBuilder.header(GRPC_STATUS, grpcStatus);
+    }
+    String grpcMessage = trailers.get(GRPC_MESSAGE);
+    if (grpcMessage != null) {
+      responseBuilder.header(GRPC_MESSAGE, grpcMessage);
     }
-    return RetryUtil.retryableGrpcStatusCodes().contains(grpcStatus);
+    return responseBuilder.build();
   }
 
   // From grpc-java

exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptor.java

@@ -38,6 +38,7 @@ public final class RetryInterceptor implements Interceptor {
   private final RetryPolicy retryPolicy;
   private final Function<Response, Boolean> isRetryable;
   private final Function<Response, OptionalLong> retryDelayNanosExtractor;
+  private final ResponseTransformer responseTransformer;
   private final Predicate<IOException> retryExceptionPredicate;
   private final Sleeper sleeper;
   private final Supplier<Double> randomJitter;
@@ -55,7 +56,26 @@ public RetryInterceptor(
             ? RetryInterceptor::isRetryableException
             : retryPolicy.getRetryExceptionPredicate(),
         TimeUnit.NANOSECONDS::sleep,
-        () -> ThreadLocalRandom.current().nextDouble(0.8d, 1.2d));
+        () -> ThreadLocalRandom.current().nextDouble(0.8d, 1.2d),
+        response -> response);
+  }
+
+  // Visible for testing
+  RetryInterceptor(
+      RetryPolicy retryPolicy,
+      Function<Response, Boolean> isRetryable,
+      Function<Response, OptionalLong> retryDelayNanosExtractor,
+      ResponseTransformer responseTransformer) {
+    this(
+        retryPolicy,
+        isRetryable,
+        retryDelayNanosExtractor,
+        retryPolicy.getRetryExceptionPredicate() == null
+            ? RetryInterceptor::isRetryableException
+            : retryPolicy.getRetryExceptionPredicate(),
+        TimeUnit.NANOSECONDS::sleep,
+        () -> ThreadLocalRandom.current().nextDouble(0.8d, 1.2d),
+        responseTransformer);
   }
 
   // Visible for testing
@@ -66,12 +86,32 @@ public RetryInterceptor(
       Predicate<IOException> retryExceptionPredicate,
       Sleeper sleeper,
       Supplier<Double> randomJitter) {
+    this(
+        retryPolicy,
+        isRetryable,
+        retryDelayNanosExtractor,
+        retryExceptionPredicate,
+        sleeper,
+        randomJitter,
+        response -> response);
+  }
+
+  // Visible for testing
+  RetryInterceptor(
+      RetryPolicy retryPolicy,
+      Function<Response, Boolean> isRetryable,
+      Function<Response, OptionalLong> retryDelayNanosExtractor,
+      Predicate<IOException> retryExceptionPredicate,
+      Sleeper sleeper,
+      Supplier<Double> randomJitter,
+      ResponseTransformer responseTransformer) {
     this.retryPolicy = retryPolicy;
     this.isRetryable = isRetryable;
     this.retryDelayNanosExtractor = retryDelayNanosExtractor;
     this.retryExceptionPredicate = retryExceptionPredicate;
     this.sleeper = sleeper;
     this.randomJitter = randomJitter;
+    this.responseTransformer = responseTransformer;
   }
 
   @Override
@@ -108,6 +148,7 @@ public Response intercept(Chain chain) throws IOException {
       try {
         response = chain.proceed(chain.request());
         if (response != null) {
+          response = responseTransformer.transform(response);
           boolean retryable = Boolean.TRUE.equals(isRetryable.apply(response));
           if (logger.isLoggable(Level.FINER)) {
             logger.log(
@@ -152,6 +193,11 @@ public Response intercept(Chain chain) throws IOException {
     throw exception;
   }
 
+  @FunctionalInterface
+  interface ResponseTransformer {
+    Response transform(Response response) throws IOException;
+  }
+
   private static String responseStringRepresentation(Response response) {
     StringJoiner joiner = new StringJoiner(",", "Response{", "}");
     joiner.add("code=" + response.code());

exporters/sender/okhttp/src/test/java/io/opentelemetry/exporter/sender/okhttp/internal/OkHttpGrpcSenderTest.java

@@ -32,6 +32,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLException;
+import okhttp3.Headers;
 import okhttp3.MediaType;
 import okhttp3.Protocol;
 import okhttp3.Request;
@@ -67,6 +68,22 @@ void isRetryable_NonRetryableGrpcStatus() {
     assertFalse(isRetryable);
   }
 
+  @Test
+  void isRetryable_RetryableGrpcStatusInTrailers() throws IOException {
+    Response response =
+        new Response.Builder()
+            .request(new Request.Builder().url("http://localhost/").build())
+            .protocol(Protocol.HTTP_2)
+            .code(200)
+            .body(ResponseBody.create("", TEXT_PLAIN))
+            .message("Retryable")
+            .trailers(() -> Headers.of(GRPC_STATUS, "14"))
+            .build();
+
+    assertTrue(OkHttpGrpcSender.isRetryable(response));
+    assertThat(response.body().string()).isEmpty();
+  }
+
   @Test
   void send_rejectedExecution_callsOnError() {
     ThreadPoolExecutor executor =