Return HTTP 400 with informative message for client-side protobuf validation errors (#6989)

indrajababu · web-flow · commit 066486c0615b · 2026-07-18T10:41:43.000-07:00
When a client sends a protobuf request containing invalid UTF-8 in string fields
(e.g., InstrumentationScope name/version), the pipeline currently returns HTTP 500
with a generic error. This change returns HTTP 400 with a client-actionable message
explaining the issue.

Changes:
- Override apply() in GrpcRequestExceptionHandler to intercept
  InvalidProtocolBufferException before Armeria's default short-circuit
- Detect via stable cause-chain type check (not brittle message string)
- Apply same fix to HTTP path handlers in otel-logs-source and otel-trace-source
- Add wire-level integration test proving HTTP 400 for invalid UTF-8 protobuf
- Add test for unsupported grpc-encoding header rejection

Signed-off-by: Indraja Babu &lt;indbabu@amazon.com&gt;
diff --git a/data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/GrpcRequestExceptionHandler.java b/data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/GrpcRequestExceptionHandler.java
@@ -13,10 +13,10 @@
 import com.linecorp.armeria.common.RequestContext;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.grpc.GoogleGrpcExceptionHandlerFunction;
+import com.linecorp.armeria.common.util.Exceptions;
 import com.linecorp.armeria.server.RequestTimeoutException;
 import io.grpc.Metadata;
 import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
 import io.micrometer.core.instrument.Counter;
 
 import org.opensearch.dataprepper.exceptions.BadRequestException;
@@ -53,6 +53,25 @@ public GrpcRequestExceptionHandler(final PluginMetrics pluginMetrics, Duration r
         retryInfoCalculator = new RetryInfoCalculator(retryInfoMinDelay, retryInfoMaxDelay);
     }
 
+    @Override
+    public Status apply(final RequestContext ctx, final Status status, final Throwable throwable, final Metadata metadata) {
+        // Armeria's GoogleGrpcExceptionHandlerFunction returns the Status of any StatusRuntimeException/
+        // StatusException verbatim and never invokes applyStatusProto(...) for it. Protobuf deserialization
+        // failures surface as StatusRuntimeException(INTERNAL, ...) wrapping an InvalidProtocolBufferException,
+        // so without this interception the client receives an opaque HTTP 500 for what is actually a client-side
+        // data problem. Remap these to an informative INVALID_ARGUMENT (HTTP 400) before delegating to the default.
+        final Throwable cause = Exceptions.peel(throwable);
+        if (InvalidRequestExceptions.isInvalidProtobuf(cause)) {
+            badRequestsCounter.increment();
+            return Status.INVALID_ARGUMENT.withDescription(InvalidRequestExceptions.INVALID_PROTOBUF_MESSAGE).withCause(cause);
+        }
+        if (InvalidRequestExceptions.isUndecodableCompressedFrame(cause)) {
+            badRequestsCounter.increment();
+            return Status.INVALID_ARGUMENT.withDescription(InvalidRequestExceptions.COMPRESSED_FRAME_MESSAGE).withCause(cause);
+        }
+        return GoogleGrpcExceptionHandlerFunction.super.apply(ctx, status, throwable, metadata);
+    }
+
     @Override
     public com.google.rpc.@Nullable Status applyStatusProto(RequestContext ctx, Throwable throwable,
                                                             Metadata metadata) {
@@ -61,7 +80,6 @@ public GrpcRequestExceptionHandler(final PluginMetrics pluginMetrics, Duration r
     }
 
     private com.google.rpc.Status handleExceptions(final Throwable e) {
-        String message = e.getMessage();
         if (e instanceof RequestTimeoutException || e instanceof TimeoutException) {
             requestTimeoutsCounter.increment();
             return createStatus(e, Status.Code.RESOURCE_EXHAUSTED);
@@ -71,9 +89,6 @@ private com.google.rpc.Status handleExceptions(final Throwable e) {
         } else if (e instanceof BadRequestException) {
             badRequestsCounter.increment();
             return createStatus(e, Status.Code.INVALID_ARGUMENT);
-        } else if ((e instanceof StatusRuntimeException) && (message.contains("Invalid protobuf byte sequence") || message.contains("Can't decode compressed frame"))) {
-            badRequestsCounter.increment();
-            return createStatus(e, Status.Code.INVALID_ARGUMENT);
         } else if (e instanceof RequestCancelledException) {
             requestTimeoutsCounter.increment();
             return createStatus(e, Status.Code.CANCELLED);
@@ -85,12 +100,19 @@ private com.google.rpc.Status handleExceptions(final Throwable e) {
     }
 
     private com.google.rpc.Status createStatus(final Throwable e, final Status.Code code) {
-        com.google.rpc.Status.Builder builder = com.google.rpc.Status.newBuilder().setCode(code.value());
+        final String message;
         if (e instanceof RequestTimeoutException) {
-            builder.setMessage(ARMERIA_REQUEST_TIMEOUT_MESSAGE);
+            message = ARMERIA_REQUEST_TIMEOUT_MESSAGE;
         } else {
-            builder.setMessage(e.getMessage() == null ? code.name() :e.getMessage());
+            message = e.getMessage() == null ? code.name() : e.getMessage();
         }
+        return createStatus(code, message);
+    }
+
+    private com.google.rpc.Status createStatus(final Status.Code code, final String message) {
+        com.google.rpc.Status.Builder builder = com.google.rpc.Status.newBuilder()
+                .setCode(code.value())
+                .setMessage(message);
         if (code == Status.Code.RESOURCE_EXHAUSTED) {
             builder.addDetails(Any.pack(retryInfoCalculator.createRetryInfo()));
         }
diff --git a/data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/InvalidRequestExceptions.java b/data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/InvalidRequestExceptions.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+
+/**
+ * Classifies request-level failures that represent client-side data problems (as opposed to server
+ * errors) so sources can respond with an informative 4xx instead of an opaque 500.
+ *
+ * <p>Detection walks the exception cause chain and keys off the stable, public
+ * {@link InvalidProtocolBufferException} type rather than an Armeria-internal error message, so it does
+ * not silently regress when the underlying framework rewords its messages.
+ */
+public final class InvalidRequestExceptions {
+
+    /**
+     * Marker used by Armeria when a compressed request frame cannot be decoded. Decompression failures do
+     * not wrap an {@link InvalidProtocolBufferException}, so this remains a message-based fallback.
+     */
+    static final String COMPRESSED_FRAME_MARKER = "Can't decode compressed frame";
+
+    public static final String INVALID_PROTOBUF_MESSAGE =
+            "Invalid protobuf byte sequence: the request payload could not be parsed as valid protobuf. "
+                    + "This is a problem with the client-side request or a misconfiguration of the pipeline source. Protobuf requires every string field to "
+                    + "contain valid UTF-8. The request likely has a field such as the InstrumentationScope name or version, a resource/scope "
+                    + "attribute key, or an attribute value that contains non-UTF-8 bytes. Check the instrumentation "
+                    + "libraries producing this telemetry and correct the source emitting the malformed data.";
+
+    public static final String COMPRESSED_FRAME_MESSAGE =
+            "Unable to decode the compressed request frame. This is a client-side data issue, not a server error. "
+                    + "Verify that the payload compression (for example, gzip) matches the compression configured on the "
+                    + "pipeline source and that the payload is not truncated or corrupted.";
+
+    private static final int MAX_CAUSE_DEPTH = 20;
+
+    private InvalidRequestExceptions() {
+    }
+
+    /**
+     * @return {@code true} if the throwable, or any cause within it, is a protobuf parse failure such as
+     *         invalid UTF-8 in a string field.
+     */
+    public static boolean isInvalidProtobuf(final Throwable throwable) {
+        Throwable current = throwable;
+        for (int depth = 0; current != null && depth < MAX_CAUSE_DEPTH; depth++) {
+            if (current instanceof InvalidProtocolBufferException) {
+                return true;
+            }
+            if (current.getCause() == current) {
+                break;
+            }
+            current = current.getCause();
+        }
+        return false;
+    }
+
+    /**
+     * @return {@code true} if the throwable, or any cause within it, is a request-frame decompression failure.
+     */
+    public static boolean isUndecodableCompressedFrame(final Throwable throwable) {
+        Throwable current = throwable;
+        for (int depth = 0; current != null && depth < MAX_CAUSE_DEPTH; depth++) {
+            final String message = current.getMessage();
+            if (message != null && message.contains(COMPRESSED_FRAME_MARKER)) {
+                return true;
+            }
+            if (current.getCause() == current) {
+                break;
+            }
+            current = current.getCause();
+        }
+        return false;
+    }
+}
diff --git a/data-prepper-plugins/armeria-common/src/test/java/org/opensearch/dataprepper/GrpcRequestExceptionHandlerTest.java b/data-prepper-plugins/armeria-common/src/test/java/org/opensearch/dataprepper/GrpcRequestExceptionHandlerTest.java
@@ -168,6 +168,51 @@ public void testHandleRequestCancelledException() {
         verify(requestTimeoutsCounter, times(2)).increment();
     }
 
+    @Test
+    public void testHandleInvalidProtobufReturnsInformativeBadRequest() {
+        // Mirrors what Armeria's GrpcMessageMarshaller throws on a protobuf UTF-8 failure:
+        // Status.INTERNAL.withDescription("Invalid protobuf byte sequence").withCause(InvalidProtocolBufferException).asRuntimeException()
+        final com.google.protobuf.InvalidProtocolBufferException protobufException =
+                new com.google.protobuf.InvalidProtocolBufferException("Protocol message had invalid UTF-8.");
+        final io.grpc.StatusRuntimeException invalidProtobuf = Status.INTERNAL
+                .withDescription("Invalid protobuf byte sequence")
+                .withCause(protobufException)
+                .asRuntimeException();
+
+        final Status resultStatus = grpcRequestExceptionHandler.apply(requestContext, Status.INTERNAL, invalidProtobuf, metadata);
+
+        assertThat(resultStatus.getCode(), equalTo(Status.Code.INVALID_ARGUMENT));
+        assertThat(resultStatus.getDescription(), equalTo(InvalidRequestExceptions.INVALID_PROTOBUF_MESSAGE));
+
+        verify(badRequestsCounter, times(1)).increment();
+    }
+
+    @Test
+    public void testHandleCompressedFrameReturnsInformativeBadRequest() {
+        final io.grpc.StatusRuntimeException compressedFrame =
+                Status.INTERNAL.withDescription("Can't decode compressed frame").asRuntimeException();
+
+        final Status resultStatus = grpcRequestExceptionHandler.apply(requestContext, Status.INTERNAL, compressedFrame, metadata);
+
+        assertThat(resultStatus.getCode(), equalTo(Status.Code.INVALID_ARGUMENT));
+        assertThat(resultStatus.getDescription(), equalTo(InvalidRequestExceptions.COMPRESSED_FRAME_MESSAGE));
+
+        verify(badRequestsCounter, times(1)).increment();
+    }
+
+    @Test
+    public void testHandleGenericStatusRuntimeExceptionPassesThroughUnchanged() {
+        // A StatusRuntimeException that is NOT a client-data problem must retain its original status,
+        // matching Armeria's default short-circuit behavior.
+        final io.grpc.StatusRuntimeException generic =
+                Status.INTERNAL.withDescription("some internal failure").asRuntimeException();
+
+        final Status resultStatus = grpcRequestExceptionHandler.apply(requestContext, Status.INTERNAL, generic, metadata);
+
+        assertThat(resultStatus.getCode(), equalTo(Status.Code.INTERNAL));
+        verify(badRequestsCounter, times(0)).increment();
+    }
+
     @Test
     public void testHandleInternalServerException() {
         final RuntimeException runtimeExceptionNoMessage = new RuntimeException();
diff --git a/data-prepper-plugins/otel-logs-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/HttpExceptionHandler.java b/data-prepper-plugins/otel-logs-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/HttpExceptionHandler.java
@@ -14,6 +14,7 @@
 import java.time.Duration;
 import java.util.concurrent.TimeoutException;
 
+import org.opensearch.dataprepper.InvalidRequestExceptions;
 import org.opensearch.dataprepper.RetryInfoCalculator;
 import org.opensearch.dataprepper.exceptions.BadRequestException;
 import org.opensearch.dataprepper.exceptions.BufferWriteException;
@@ -38,7 +39,6 @@
 import com.linecorp.armeria.server.annotation.ExceptionHandlerFunction;
 
 import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
 import io.micrometer.core.instrument.Counter;
 
 public class HttpExceptionHandler implements ExceptionHandlerFunction {
@@ -101,9 +101,14 @@ private StatusHolder createStatus(Throwable e) {
         } else if (e instanceof BadRequestException) {
             badRequestsCounter.increment();
             return new StatusHolder(createStatus(e, Status.Code.INVALID_ARGUMENT), createHttpStatusFromProtoBufStatus(Status.Code.INVALID_ARGUMENT));
-        } else if ((e instanceof StatusRuntimeException) && (e.getMessage().contains("Invalid protobuf byte sequence") || e.getMessage().contains("Can't decode compressed frame"))) {
+        } else if (InvalidRequestExceptions.isInvalidProtobuf(e)) {
             badRequestsCounter.increment();
-            return new StatusHolder(createStatus(e, Status.Code.INVALID_ARGUMENT), createHttpStatusFromProtoBufStatus(Status.Code.INVALID_ARGUMENT));
+            return new StatusHolder(createStatus(Status.Code.INVALID_ARGUMENT, InvalidRequestExceptions.INVALID_PROTOBUF_MESSAGE),
+                    createHttpStatusFromProtoBufStatus(Status.Code.INVALID_ARGUMENT));
+        } else if (InvalidRequestExceptions.isUndecodableCompressedFrame(e)) {
+            badRequestsCounter.increment();
+            return new StatusHolder(createStatus(Status.Code.INVALID_ARGUMENT, InvalidRequestExceptions.COMPRESSED_FRAME_MESSAGE),
+                    createHttpStatusFromProtoBufStatus(Status.Code.INVALID_ARGUMENT));
         } else if (e instanceof RequestCancelledException) {
             requestTimeoutsCounter.increment();
             return new StatusHolder(createStatus(e, Status.Code.CANCELLED), createHttpStatusFromProtoBufStatus(Status.Code.CANCELLED));
@@ -125,12 +130,16 @@ private HttpStatus createHttpStatusFromProtoBufStatus(Status.Code status) {
     }
 
     private com.google.rpc.Status createStatus(final Throwable e, final Status.Code code) {
-        com.google.rpc.Status.Builder builder = com.google.rpc.Status.newBuilder().setCode(code.value());
-        if (e instanceof RequestTimeoutException) {
-            builder.setMessage(ARMERIA_REQUEST_TIMEOUT_MESSAGE);
-        } else {
-            builder.setMessage(e.getMessage() == null ? code.name() :e.getMessage());
-        }
+        final String message = e instanceof RequestTimeoutException
+                ? ARMERIA_REQUEST_TIMEOUT_MESSAGE
+                : (e.getMessage() == null ? code.name() : e.getMessage());
+        return createStatus(code, message);
+    }
+
+    private com.google.rpc.Status createStatus(final Status.Code code, final String message) {
+        com.google.rpc.Status.Builder builder = com.google.rpc.Status.newBuilder()
+                .setCode(code.value())
+                .setMessage(message);
         if (code == Status.Code.RESOURCE_EXHAUSTED) {
             builder.addDetails(Any.pack(retryInfoCalculator.createRetryInfo()));
         }
diff --git a/data-prepper-plugins/otel-logs-source/src/test/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSourceUnframedRequestTest.java b/data-prepper-plugins/otel-logs-source/src/test/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsSourceUnframedRequestTest.java
@@ -10,9 +10,11 @@
 
 package org.opensearch.dataprepper.plugins.source.otellogs;
 
+import static com.linecorp.armeria.common.HttpStatus.BAD_REQUEST;
 import static com.linecorp.armeria.common.HttpStatus.OK;
 import static com.linecorp.armeria.common.HttpStatus.UNSUPPORTED_MEDIA_TYPE;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.is;
@@ -169,6 +171,56 @@ void unframedRequests_unframeRequestsAreDisabledAndHttpRequestIsSentToGrpcEndpoi
                 .join();
     }
 
+    @Test
+    void unframedRequests_protobufWithInvalidUtf8_returns400WithInformativeMessage() {
+        configureSource(createDefaultConfigBuilder().enableUnframedRequests(true).build());
+        SOURCE.start(buffer);
+
+        // Hand-crafted ExportLogsServiceRequest wire bytes whose nested InstrumentationScope.name (field 1)
+        // contains a lone 0xFF byte, which is invalid UTF-8. Protobuf's String fields require valid UTF-8, so
+        // deserialization fails exactly as it did for the customer. Structure:
+        // ExportLogsServiceRequest{ resource_logs[0]{ scope_logs[0]{ scope{ name = 0xFF } } } }
+        final byte[] invalidUtf8Protobuf = new byte[] {
+                0x0A, 0x07,             // ExportLogsServiceRequest.resource_logs (field 1, len 7)
+                0x12, 0x05,             //   ResourceLogs.scope_logs (field 2, len 5)
+                0x0A, 0x03,             //     ScopeLogs.scope (field 1, len 3)
+                0x0A, 0x01, (byte) 0xFF //       InstrumentationScope.name (field 1, len 1) = 0xFF (invalid UTF-8)
+        };
+
+        final AggregatedHttpResponse response = WebClient.of().execute(getDefaultRequestHeadersBuilder()
+                                .path(CONFIG_GRPC_PATH)
+                                .contentType(MediaType.PROTOBUF)
+                                .build(),
+                        HttpData.wrap(invalidUtf8Protobuf))
+                .aggregate()
+                .join();
+
+        assertThat("Invalid UTF-8 protobuf must be rejected as a client error, not a 500",
+                response.status(), equalTo(BAD_REQUEST));
+        final String grpcMessage = response.headers().get("grpc-message");
+        final String detail = grpcMessage != null ? grpcMessage : response.contentUtf8();
+        assertThat("Client should receive an informative, client-side-oriented message",
+                detail, containsString("contain valid UTF-8"));
+    }
+
+    @Test
+    void unframedRequests_requestWithUnsupportedGrpcEncoding_isRejected() {
+        configureSource(createDefaultConfigBuilder().enableUnframedRequests(true).build());
+        SOURCE.start(buffer);
+
+        final AggregatedHttpResponse response = WebClient.of().execute(getDefaultRequestHeadersBuilder()
+                                .path(CONFIG_GRPC_PATH)
+                                .contentType(MediaType.PROTOBUF)
+                                .add("grpc-encoding", "gzip")
+                                .build(),
+                        HttpData.wrap(new byte[] {0x00}))
+                .aggregate()
+                .join();
+
+        assertThat("A request the pipeline source cannot decode must not be a 500",
+                response.status().isServerError(), equalTo(false));
+    }
+
     private void assertSecureResponseWithStatusCode(final AggregatedHttpResponse response,
                                                     final HttpStatus expectedStatus,
                                                     final Throwable throwable) {
diff --git a/data-prepper-plugins/otel-trace-source/src/main/java/org/opensearch/dataprepper/plugins/source/oteltrace/http/HttpExceptionHandler.java b/data-prepper-plugins/otel-trace-source/src/main/java/org/opensearch/dataprepper/plugins/source/oteltrace/http/HttpExceptionHandler.java
