This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 856f766b204 perf: RemoteTaskActionClient streaming deserialization.
(#20005)
856f766b204 is described below
commit 856f766b204fab1f016f7d98da3073ef10902a61
Author: Gian Merlino <[email protected]>
AuthorDate: Fri Aug 14 10:15:27 2026 -0700
perf: RemoteTaskActionClient streaming deserialization. (#20005)
This patch switches RemoteTaskActionClient to use InputStreamResponseHandler
rather than BytesFullResponseHandler, which enables deserialization
to run without buffering the entire response. This is useful for actions
with potentially large responses, such as RetrieveUsedSegmentsAction.
---
.../common/actions/RemoteTaskActionClient.java | 41 +++++++++++++++-------
.../common/actions/RemoteTaskActionClientTest.java | 17 ++++-----
2 files changed, 35 insertions(+), 23 deletions(-)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RemoteTaskActionClient.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RemoteTaskActionClient.java
index 4015b99558a..9b6e8edbf72 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RemoteTaskActionClient.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RemoteTaskActionClient.java
@@ -19,12 +19,13 @@
package org.apache.druid.indexing.common.actions;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.java.util.common.IOE;
-import org.apache.druid.java.util.common.jackson.JacksonUtils;
import org.apache.druid.java.util.common.logger.Logger;
-import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
+import
org.apache.druid.java.util.http.client.response.InputStreamResponseHandler;
import
org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
import org.apache.druid.rpc.HttpResponseException;
import org.apache.druid.rpc.RequestBuilder;
@@ -32,7 +33,6 @@ import org.apache.druid.rpc.ServiceClient;
import org.jboss.netty.handler.codec.http.HttpMethod;
import java.io.IOException;
-import java.util.Map;
import java.util.concurrent.ExecutionException;
public class RemoteTaskActionClient implements TaskActionClient
@@ -63,19 +63,17 @@ public class RemoteTaskActionClient implements
TaskActionClient
// We're using a ServiceClient directly here instead of OverlordClient,
because OverlordClient does
// not have access to the TaskAction class. (OverlordClient is in the
druid-server package, and TaskAction
// is in the druid-indexing-service package.)
- final Map<String, Object> response = jsonMapper.readValue(
+ return jsonMapper.<TaskActionResponse<RetType>>readValue(
client.request(
new RequestBuilder(HttpMethod.POST, "/druid/indexer/v1/action")
.jsonContent(jsonMapper, new TaskActionHolder(task,
taskAction)),
- new BytesFullResponseHandler()
- ).getContent(),
- JacksonUtils.TYPE_REFERENCE_MAP_STRING_OBJECT
- );
-
- return jsonMapper.convertValue(
- response.get("result"),
- taskAction.getReturnTypeReference()
- );
+ new InputStreamResponseHandler()
+ ),
+ jsonMapper.getTypeFactory().constructParametricType(
+ TaskActionResponse.class,
+
jsonMapper.getTypeFactory().constructType(taskAction.getReturnTypeReference())
+ )
+ ).getResult();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -95,4 +93,21 @@ public class RemoteTaskActionClient implements
TaskActionClient
throw new IOException(e.getCause());
}
}
+
+ public static class TaskActionResponse<RetType>
+ {
+ private final RetType result;
+
+ @JsonCreator
+ public TaskActionResponse(@JsonProperty("result") RetType result)
+ {
+ this.result = result;
+ }
+
+ @JsonProperty
+ public RetType getResult()
+ {
+ return result;
+ }
+ }
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RemoteTaskActionClientTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RemoteTaskActionClientTest.java
index a84fe68ab2b..3f77fc04a5d 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RemoteTaskActionClientTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RemoteTaskActionClientTest.java
@@ -28,8 +28,7 @@ import org.apache.druid.indexing.common.task.NoopTask;
import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.Intervals;
-import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
-import org.apache.druid.java.util.http.client.response.BytesFullResponseHolder;
+import
org.apache.druid.java.util.http.client.response.InputStreamResponseHandler;
import
org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
import org.apache.druid.rpc.HttpResponseException;
import org.apache.druid.rpc.RequestBuilder;
@@ -38,15 +37,14 @@ import org.apache.druid.rpc.ServiceClientImpl;
import org.apache.druid.rpc.StandardRetryPolicy;
import org.easymock.EasyMock;
import org.jboss.netty.buffer.BigEndianHeapChannelBuffer;
-import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.jboss.netty.handler.codec.http.HttpResponse;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
-import org.jboss.netty.handler.codec.http.HttpVersion;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
@@ -81,9 +79,8 @@ public class RemoteTaskActionClientTest
));
expectedResponse.put("result", expectedLocks);
- final DefaultHttpResponse httpResponse = new
DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
- final BytesFullResponseHolder responseHolder = new
BytesFullResponseHolder(httpResponse);
- responseHolder.addChunk(objectMapper.writeValueAsBytes(expectedResponse));
+ final ByteArrayInputStream responseStream =
+ new
ByteArrayInputStream(objectMapper.writeValueAsBytes(expectedResponse));
final Task task = NoopTask.create();
final LockListAction action = new LockListAction();
@@ -93,10 +90,10 @@ public class RemoteTaskActionClientTest
EasyMock.eq(
new RequestBuilder(HttpMethod.POST,
"/druid/indexer/v1/action")
.jsonContent(objectMapper, new
TaskActionHolder(task, action))),
- EasyMock.anyObject(BytesFullResponseHandler.class)
+ EasyMock.anyObject(InputStreamResponseHandler.class)
)
)
- .andReturn(responseHolder);
+ .andReturn(responseStream);
EasyMock.replay(directOverlordClient);
@@ -129,7 +126,7 @@ public class RemoteTaskActionClientTest
new RequestBuilder(HttpMethod.POST,
"/druid/indexer/v1/action")
.jsonContent(objectMapper, new
TaskActionHolder(task, action))
),
- EasyMock.anyObject(BytesFullResponseHandler.class)
+ EasyMock.anyObject(InputStreamResponseHandler.class)
)
)
.andThrow(new ExecutionException(new
HttpResponseException(responseHolder)));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
