This is an automated email from the ASF dual-hosted git repository.
capistrant pushed a commit to branch 39.0.0
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/39.0.0 by this push:
new 21e8f6409ea fix!: send query response trailers only to clients that
send TE: trailers (#20717) (#20751)
21e8f6409ea is described below
commit 21e8f6409ea8328c839015d7bfd69ba8693c05f6
Author: Lucas Capistrant <[email protected]>
AuthorDate: Tue Oct 27 20:32:22 2026 -0500
fix!: send query response trailers only to clients that send TE: trailers
(#20717) (#20751)
JDK 17.0.21, 21.0.13 and 25.0.5 (JDK-8383439) make java.net.http.HttpClient
reject chunked responses with a trailer section (ProtocolException: CR
expected). Brokers and Historicals always sent X-Druid-Response-Complete as
a
trailer, so streamed query responses failed for those clients. Only declare
and send trailers when the request has TE: trailers (RFC 9110 10.1.4).
Co-authored-by: Frank Chen <[email protected]>
---
.../embedded/server/QueryResponseTrailerTest.java | 188 +++++++++++++++++++++
.../org/apache/druid/server/QueryResultPusher.java | 39 ++++-
.../org/apache/druid/server/QueryResourceTest.java | 14 ++
.../apache/druid/server/QueryResultPusherTest.java | 23 +++
.../druid/server/mocks/MockHttpServletRequest.java | 4 +-
5 files changed, 261 insertions(+), 7 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/server/QueryResponseTrailerTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/server/QueryResponseTrailerTest.java
new file mode 100644
index 00000000000..afaf40f4344
--- /dev/null
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/server/QueryResponseTrailerTest.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.druid.testing.embedded.server;
+
+import org.apache.druid.server.QueryResource;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedDruidServer;
+import org.apache.druid.testing.embedded.EmbeddedHistorical;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.InputStream;
+import java.net.Socket;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Locale;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Verifies that streamed query responses carry the {@link
QueryResource#RESPONSE_COMPLETE_TRAILER_HEADER} trailer only
+ * when the client sends {@code TE: trailers}, so that clients which reject
trailers, such as
+ * {@link java.net.http.HttpClient} on recent JDK updates, can read query
results.
+ */
+public class QueryResponseTrailerTest extends EmbeddedClusterTestBase
+{
+ private static final String SQL_QUERY = "{\"query\": \"SELECT 1\"}";
+ private static final String NATIVE_QUERY = """
+ {
+ "queryType": "timeseries",
+ "dataSource": "missing_datasource",
+ "granularity": "all",
+ "intervals": ["2000/3000"],
+ "aggregations": [{"type": "count", "name": "rows"}]
+ }
+ """;
+
+ private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator();
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+ private final EmbeddedHistorical historical = new EmbeddedHistorical();
+ private final HttpClient client = HttpClient.newHttpClient();
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper()
+ .addServer(coordinator)
+ .addServer(overlord)
+ .addServer(broker)
+ .addServer(historical);
+ }
+
+ @AfterAll
+ public void closeClient()
+ {
+ client.close();
+ }
+
+ @Test
+ @Timeout(30)
+ public void testTrailersOnlyWhenClientSendsTeTrailers() throws Exception
+ {
+ for (final String path : new String[]{"/druid/v2/sql", "/druid/v2"}) {
+ final String body = path.endsWith("sql") ? SQL_QUERY : NATIVE_QUERY;
+
+ final String withoutTe = sendRaw(broker, path, body, false);
+ Assertions.assertTrue(withoutTe.startsWith("HTTP/1.1 200"), withoutTe);
+
Assertions.assertFalse(withoutTe.contains(QueryResource.RESPONSE_COMPLETE_TRAILER_HEADER),
withoutTe);
+
+ final String withTe = sendRaw(broker, path, body, true);
+ Assertions.assertTrue(withTe.startsWith("HTTP/1.1 200"), withTe);
+ Assertions.assertTrue(
+ withTe.endsWith("\r\n0\r\n" +
QueryResource.RESPONSE_COMPLETE_TRAILER_HEADER + ": true\r\n\r\n"),
+ withTe
+ );
+ }
+ }
+
+ @Test
+ @Timeout(30)
+ public void testJavaHttpClientReadsStreamedResults() throws Exception
+ {
+ assertJavaHttpClientReadsResults(broker, "/druid/v2/sql", SQL_QUERY);
+ assertJavaHttpClientReadsResults(broker, "/druid/v2", NATIVE_QUERY);
+ assertJavaHttpClientReadsResults(historical, "/druid/v2", NATIVE_QUERY);
+ }
+
+ private void assertJavaHttpClientReadsResults(
+ final EmbeddedDruidServer<?> server,
+ final String path,
+ final String body
+ ) throws Exception
+ {
+ final HttpRequest request =
HttpRequest.newBuilder(URI.create(getServerUrl(server) + path))
+ .header("Content-Type",
"application/json")
+ .timeout(Duration.ofSeconds(10))
+
.POST(HttpRequest.BodyPublishers.ofString(body))
+ .build();
+ final HttpResponse<InputStream> response = client.send(request,
HttpResponse.BodyHandlers.ofInputStream());
+ Assertions.assertEquals(200, response.statusCode());
+ try (InputStream in = response.body()) {
+ Assertions.assertTrue(new String(in.readAllBytes(),
StandardCharsets.UTF_8).startsWith("["));
+ }
+ }
+
+ /**
+ * Sends a request over a raw keep-alive socket and returns the response
bytes as they appear on the wire, including
+ * chunk framing and trailers, which {@link HttpClient} does not expose.
+ */
+ private String sendRaw(
+ final EmbeddedDruidServer<?> server,
+ final String path,
+ final String body,
+ final boolean teTrailers
+ ) throws Exception
+ {
+ final URI uri = URI.create(getServerUrl(server));
+ final byte[] bodyBytes = body.getBytes(StandardCharsets.UTF_8);
+ final String request = "POST " + path + " HTTP/1.1\r\n"
+ + "Host: " + uri.getHost() + "\r\n"
+ + "Content-Type: application/json\r\n"
+ + "Content-Length: " + bodyBytes.length + "\r\n"
+ + (teTrailers ? "TE: trailers\r\nConnection:
TE\r\n" : "")
+ + "\r\n";
+ try (Socket socket = new Socket(uri.getHost(), uri.getPort())) {
+ socket.setSoTimeout(10_000);
+
socket.getOutputStream().write(request.getBytes(StandardCharsets.US_ASCII));
+ socket.getOutputStream().write(bodyBytes);
+ socket.getOutputStream().flush();
+
+ final InputStream in = socket.getInputStream();
+ final StringBuilder response = new StringBuilder();
+ while (!isComplete(response)) {
+ final int b = in.read();
+ if (b < 0) {
+ break;
+ }
+ response.append((char) b);
+ }
+ return response.toString();
+ }
+ }
+
+ private static boolean isComplete(final CharSequence response)
+ {
+ final String text = response.toString();
+ final int headersEnd = text.indexOf("\r\n\r\n");
+ if (headersEnd < 0) {
+ return false;
+ }
+ final String headers = text.substring(0,
headersEnd).toLowerCase(Locale.ROOT);
+ final Matcher contentLength = Pattern.compile("\r\ncontent-length:
*(\\d+)").matcher(headers);
+ if (contentLength.find()) {
+ return text.length() >= headersEnd + 4 +
Integer.parseInt(contentLength.group(1));
+ }
+ // Chunked: complete once the last chunk and the (possibly empty) trailer
section have been read.
+ final int lastChunk = text.indexOf("\r\n0\r\n", headersEnd);
+ return lastChunk >= 0 && text.endsWith("\r\n\r\n") && text.length() >=
lastChunk + 7;
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/server/QueryResultPusher.java
b/server/src/main/java/org/apache/druid/server/QueryResultPusher.java
index 6f0b3ee272a..b788f2f47ca 100644
--- a/server/src/main/java/org/apache/druid/server/QueryResultPusher.java
+++ b/server/src/main/java/org/apache/druid/server/QueryResultPusher.java
@@ -51,6 +51,7 @@ import javax.ws.rs.core.Response;
import java.io.Closeable;
import java.io.IOException;
import java.io.OutputStream;
+import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
@@ -129,8 +130,10 @@ public abstract class QueryResultPusher
final Response.ResponseBuilder startResponse = resultsWriter.start();
if (startResponse != null) {
- startResponse.header(QueryResource.QUERY_ID_RESPONSE_HEADER, queryId)
- .header(HttpHeader.TRAILER.toString(),
RESULT_TRAILER_HEADERS);
+ startResponse.header(QueryResource.QUERY_ID_RESPONSE_HEADER, queryId);
+ if (clientAcceptsTrailers(request)) {
+ startResponse.header(HttpHeader.TRAILER.toString(),
RESULT_TRAILER_HEADERS);
+ }
for (Map.Entry<String, String> entry : extraHeaders.entrySet()) {
startResponse.header(entry.getKey(), entry.getValue());
@@ -154,8 +157,10 @@ public abstract class QueryResultPusher
response.setHeader(entry.getKey(), entry.getValue());
}
- response.setHeader(HttpHeader.TRAILER.toString(),
RESULT_TRAILER_HEADERS);
- response.setTrailerFields(() -> trailerFields);
+ if (clientAcceptsTrailers(request)) {
+ response.setHeader(HttpHeader.TRAILER.toString(),
RESULT_TRAILER_HEADERS);
+ response.setTrailerFields(() -> trailerFields);
+ }
// Start with complete status
trailerFields.put(QueryResource.RESPONSE_COMPLETE_TRAILER_HEADER,
"true");
@@ -321,6 +326,30 @@ public abstract class QueryResultPusher
}
}
+ /**
+ * Whether the client sent {@code TE: trailers}, which is how a client
signals that it accepts trailer fields
+ * (RFC 9110 section 10.1.4). Trailers are only sent to such clients,
because some clients reject a chunked response
+ * that carries trailers: {@link java.net.http.HttpClient} fails with {@code
java.net.ProtocolException: CR expected}
+ * since JDK 17.0.21, 21.0.13 and 25.0.5 (JDK-8383439).
+ */
+ static boolean clientAcceptsTrailers(final HttpServletRequest request)
+ {
+ final Enumeration<String> teHeaders =
request.getHeaders(HttpHeader.TE.asString());
+ if (teHeaders == null) {
+ return false;
+ }
+ while (teHeaders.hasMoreElements()) {
+ for (final String coding : teHeaders.nextElement().split(",")) {
+ final int paramsStart = coding.indexOf(';');
+ final String codingName = paramsStart < 0 ? coding :
coding.substring(0, paramsStart);
+ if ("trailers".equalsIgnoreCase(codingName.trim())) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
/**
* Generates a response for a {@link DruidException} that occurs prior to
any query results being sent out.
*/
@@ -448,8 +477,6 @@ public abstract class QueryResultPusher
response.setContentType(contentType.toString());
- response.setTrailerFields(() -> trailerFields);
-
try {
out = new CountingOutputStream(response.getOutputStream());
writer = resultsWriter.makeWriter(out);
diff --git
a/server/src/test/java/org/apache/druid/server/QueryResourceTest.java
b/server/src/test/java/org/apache/druid/server/QueryResourceTest.java
index 6a62b4d77c1..b37593b14ff 100644
--- a/server/src/test/java/org/apache/druid/server/QueryResourceTest.java
+++ b/server/src/test/java/org/apache/druid/server/QueryResourceTest.java
@@ -343,6 +343,7 @@ public class QueryResourceTest
public void testGoodQuery() throws IOException
{
expectPermissiveHappyPathAuth();
+ testServletRequest.headers.put(HttpHeader.TE.asString(), "trailers");
HttpServletResponse servletResponse =
expectAsyncRequestFlow(SIMPLE_TIMESERIES_QUERY);
Assertions.assertEquals(200, servletResponse.getStatus());
@@ -355,6 +356,17 @@ public class QueryResourceTest
Assertions.assertEquals(fields.get(QueryResource.RESPONSE_COMPLETE_TRAILER_HEADER),
"true");
}
+ @Test
+ public void testGoodQueryWithoutTeTrailersOmitsTrailers() throws IOException
+ {
+ expectPermissiveHappyPathAuth();
+
+ final MockHttpServletResponse servletResponse =
expectAsyncRequestFlow(SIMPLE_TIMESERIES_QUERY);
+ Assertions.assertEquals(200, servletResponse.getStatus());
+
Assertions.assertFalse(servletResponse.containsHeader(HttpHeader.TRAILER.toString()));
+ Assertions.assertNull(servletResponse.getTrailerFields());
+ }
+
@Test
public void testGoodQueryWithQueryConfigOverrideDefault() throws IOException
{
@@ -547,6 +559,7 @@ public class QueryResourceTest
);
expectPermissiveHappyPathAuth();
+ testServletRequest.headers.put(HttpHeader.TE.asString(), "trailers");
HttpServletResponse response = expectAsyncRequestFlow(testServletRequest,
SIMPLE_TIMESERIES_QUERY.getBytes(StandardCharsets.UTF_8), queryResource);
@@ -714,6 +727,7 @@ public class QueryResourceTest
);
expectPermissiveHappyPathAuth();
+ testServletRequest.headers.put(HttpHeader.TE.asString(), "trailers");
HttpServletResponse response = expectAsyncRequestFlow(testServletRequest,
SIMPLE_TIMESERIES_QUERY.getBytes(StandardCharsets.UTF_8), queryResource);
diff --git
a/server/src/test/java/org/apache/druid/server/QueryResultPusherTest.java
b/server/src/test/java/org/apache/druid/server/QueryResultPusherTest.java
index ce3e07d9bfc..ab6cb87154a 100644
--- a/server/src/test/java/org/apache/druid/server/QueryResultPusherTest.java
+++ b/server/src/test/java/org/apache/druid/server/QueryResultPusherTest.java
@@ -26,6 +26,7 @@ import
org.apache.druid.server.QueryResource.QueryMetricCounter;
import org.apache.druid.server.QueryResultPusher.ResultsWriter;
import org.apache.druid.server.QueryResultPusher.Writer;
import org.apache.druid.server.mocks.MockHttpServletRequest;
+import org.eclipse.jetty.http.HttpHeader;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -50,6 +51,28 @@ public class QueryResultPusherTest
true,
false);
+ @Test
+ public void testClientAcceptsTrailers()
+ {
+
Assertions.assertFalse(QueryResultPusher.clientAcceptsTrailers(requestWithTe(null)));
+
Assertions.assertFalse(QueryResultPusher.clientAcceptsTrailers(requestWithTe("")));
+
Assertions.assertFalse(QueryResultPusher.clientAcceptsTrailers(requestWithTe("gzip,
deflate;q=0.5")));
+
Assertions.assertFalse(QueryResultPusher.clientAcceptsTrailers(requestWithTe("trailersx")));
+
Assertions.assertTrue(QueryResultPusher.clientAcceptsTrailers(requestWithTe("trailers")));
+
Assertions.assertTrue(QueryResultPusher.clientAcceptsTrailers(requestWithTe("Trailers")));
+
Assertions.assertTrue(QueryResultPusher.clientAcceptsTrailers(requestWithTe("gzip;q=0.5,
trailers")));
+
Assertions.assertTrue(QueryResultPusher.clientAcceptsTrailers(requestWithTe("
trailers ;q=1, gzip")));
+ }
+
+ private static HttpServletRequest requestWithTe(final String te)
+ {
+ final MockHttpServletRequest request = new MockHttpServletRequest();
+ if (te != null) {
+ request.headers.put(HttpHeader.TE.asString(), te);
+ }
+ return request;
+ }
+
@Test
public void testResultPusherRetainsNestedExceptionBacktraces()
{
diff --git
a/server/src/test/java/org/apache/druid/server/mocks/MockHttpServletRequest.java
b/server/src/test/java/org/apache/druid/server/mocks/MockHttpServletRequest.java
index 0582f5871b1..ea7e9865826 100644
---
a/server/src/test/java/org/apache/druid/server/mocks/MockHttpServletRequest.java
+++
b/server/src/test/java/org/apache/druid/server/mocks/MockHttpServletRequest.java
@@ -37,6 +37,7 @@ import javax.servlet.http.Part;
import java.io.BufferedReader;
import java.security.Principal;
import java.util.Collection;
+import java.util.Collections;
import java.util.Enumeration;
import java.util.LinkedHashMap;
import java.util.Locale;
@@ -94,7 +95,8 @@ public class MockHttpServletRequest implements
HttpServletRequest
@Override
public Enumeration<String> getHeaders(String name)
{
- throw new UnsupportedOperationException();
+ final String value = headers.get(name);
+ return Collections.enumeration(value == null ? Collections.emptyList() :
Collections.singletonList(value));
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
