You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2022/06/23 05:23:25 UTC
[GitHub] [druid] gianm opened a new pull request, #12696: Mid-level service client and updated high-level clients.
gianm opened a new pull request, #12696:
URL: https://github.com/apache/druid/pull/12696
Our servers talk to each other over HTTP. We have a low-level HTTP
client (HttpClient) that is super-asynchronous and super-customizable
through its handlers. It's also proven to be quite robust: we use it
for Broker -> Historical communication over the wide variety of query
types and workloads we support.
But the low-level client has no facilities for service location or
retries, which means we have a variety of high-level clients that
implement these in their own ways. Some high-level clients do a better
job than others. This patch adds a mid-level ServiceClient that makes
it easier for high-level clients to be built correctly and harmoniously,
and migrates some of the high-level logic to use ServiceClients.
Main changes:
1) Add ServiceClient org.apache.druid.rpc package. That package also
contains supporting stuff like ServiceLocator and RetryPolicy
interfaces, and a DiscoveryServiceLocator based on
DruidNodeDiscoveryProvider.
2) Add high-level OverlordClient in org.apache.druid.rpc.indexing.
3) Indexing task client creator in TaskServiceClients. It uses
SpecificTaskServiceLocator to find the tasks. This improves on
ClientInfoTaskProvider by caching task locations for up to 30 seconds
across calls, reducing load on the Overlord.
4) Rework ParallelIndexSupervisorTaskClient to use a ServiceClient
instead of extending IndexTaskClient.
5) Rework RemoteTaskActionClient to use a ServiceClient instead of
DruidLeaderClient.
6) Rework LocalIntermediaryDataManager, TaskMonitor, and
ParallelIndexSupervisorTask. As a result, MiddleManager, Peon, and
Overlord no longer need IndexingServiceClient (which internally used
DruidLeaderClient).
There are some concrete benefits over the prior logic, namely:
- DruidLeaderClient does retries in its "go" method, but only retries
exactly 5 times, does not sleep between retries, and does not retry
retryable HTTP codes like 502, 503, 504. (It only retries IOExceptions.)
ServiceClient handles retries in a more reasonable way.
- DruidLeaderClient's methods are all synchronous, whereas ServiceClient
methods are asynchronous. This is used in one place so far: the
SpecificTaskServiceLocator, so we don't need to block a thread trying
to locate a task. It can be used in other places in the future.
- HttpIndexingServiceClient does not properly handle all server errors.
In some cases, it tries to parse a server error as a successful
response (for example: in getTaskStatus).
- IndexTaskClient currently makes an Overlord call on every task-to-task
HTTP request, as a way to find where the target task is. ServiceClient,
through SpecificTaskServiceLocator, caches these target locations
for a period of time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] samarthjain commented on a diff in pull request #12696: Mid-level service client and updated high-level clients.
Posted by GitBox <gi...@apache.org>.
samarthjain commented on code in PR #12696:
URL: https://github.com/apache/druid/pull/12696#discussion_r905731771
##########
core/src/main/java/org/apache/druid/common/guava/FutureUtils.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.common.guava;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+
+public class FutureUtils
+{
+ /**
+ * Waits for a given future and returns its value, like {@code future.get()}.
+ *
+ * On InterruptedException, cancels the provided future if {@code cancelIfInterrupted}, then re-throws the
+ * original InterruptedException.
+ *
+ * Passes through CancellationExceptions and ExecutionExceptions as-is.
+ */
+ public static <T> T get(final ListenableFuture<T> future, final boolean cancelIfInterrupted)
+ throws InterruptedException, ExecutionException
+ {
+ try {
+ return future.get();
+ }
+ catch (InterruptedException e) {
+ if (cancelIfInterrupted) {
+ future.cancel(true);
+ }
+
Review Comment:
Interrupt status probably needs to be set again by calling `Thread.currentThread().interrupt()`
##########
core/src/main/java/org/apache/druid/common/guava/FutureUtils.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.common.guava;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+
+public class FutureUtils
+{
+ /**
+ * Waits for a given future and returns its value, like {@code future.get()}.
+ *
+ * On InterruptedException, cancels the provided future if {@code cancelIfInterrupted}, then re-throws the
+ * original InterruptedException.
+ *
+ * Passes through CancellationExceptions and ExecutionExceptions as-is.
+ */
+ public static <T> T get(final ListenableFuture<T> future, final boolean cancelIfInterrupted)
+ throws InterruptedException, ExecutionException
+ {
+ try {
+ return future.get();
+ }
+ catch (InterruptedException e) {
+ if (cancelIfInterrupted) {
+ future.cancel(true);
+ }
+
+ throw e;
+ }
+ }
+
+ /**
+ * Waits for a given future and returns its value, like {@code future.get()}.
+ *
+ * On InterruptException, cancels the provided future if {@code cancelIfInterrupted}, and in either case, throws
+ * a RuntimeException that wraps the original InterruptException.
+ *
+ * Passes through CancellationExceptions as-is.
+ *
+ * Re-wraps the causes of ExecutionExceptions using RuntimeException.
+ */
+ public static <T> T getUnchecked(final ListenableFuture<T> future, final boolean cancelIfInterrupted)
+ {
+ try {
+ return FutureUtils.get(future, cancelIfInterrupted);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
Review Comment:
Interrupt status probably needs to be set again by calling Thread.currentThread().interrupt()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm merged pull request #12696: Mid-level service client and updated high-level clients.
Posted by GitBox <gi...@apache.org>.
gianm merged PR #12696:
URL: https://github.com/apache/druid/pull/12696
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on a diff in pull request #12696: Mid-level service client and updated high-level clients.
Posted by GitBox <gi...@apache.org>.
gianm commented on code in PR #12696:
URL: https://github.com/apache/druid/pull/12696#discussion_r913999234
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskClient.java:
##########
@@ -19,122 +19,28 @@
package org.apache.druid.indexing.common.task.batch.parallel;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.druid.indexing.common.IndexTaskClient;
-import org.apache.druid.indexing.common.TaskInfoProvider;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.http.client.HttpClient;
-import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
-import org.jboss.netty.handler.codec.http.HttpMethod;
import org.joda.time.DateTime;
-import org.joda.time.Duration;
import javax.annotation.Nullable;
import java.io.IOException;
-public class ParallelIndexSupervisorTaskClient extends IndexTaskClient
+public interface ParallelIndexSupervisorTaskClient
{
- ParallelIndexSupervisorTaskClient(
- HttpClient httpClient,
- ObjectMapper objectMapper,
- TaskInfoProvider taskInfoProvider,
- Duration httpTimeout,
- String callerId,
- long numRetries
- )
- {
- super(httpClient, objectMapper, taskInfoProvider, httpTimeout, callerId, 1, numRetries);
- }
-
/**
* See {@link SinglePhaseParallelIndexTaskRunner#allocateNewSegment(String, DateTime)}.
*/
@Deprecated
- public SegmentIdWithShardSpec allocateSegment(String supervisorTaskId, DateTime timestamp) throws IOException
- {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "segment/allocate",
- null,
- serialize(timestamp),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "task[%s] failed to allocate a new segment identifier with the HTTP code[%d] and content[%s]",
- supervisorTaskId,
- response.getStatus().getCode(),
- response.getContent()
- );
- } else {
- return deserialize(
- response.getContent(),
- new TypeReference<SegmentIdWithShardSpec>()
- {
- }
- );
- }
- }
+ SegmentIdWithShardSpec allocateSegment(DateTime timestamp) throws IOException;
/**
* See {@link SinglePhaseParallelIndexTaskRunner#allocateNewSegment(String, DateTime, String, String)}.
*/
- public SegmentIdWithShardSpec allocateSegment(
- String supervisorTaskId,
+ SegmentIdWithShardSpec allocateSegment(
DateTime timestamp,
String sequenceName,
@Nullable String prevSegmentId
- ) throws IOException
- {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "segment/allocate",
- null,
- serialize(new SegmentAllocationRequest(timestamp, sequenceName, prevSegmentId)),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "task[%s] failed to allocate a new segment identifier with the HTTP code[%d] and content[%s]",
- supervisorTaskId,
- response.getStatus().getCode(),
- response.getContent()
- );
- } else {
- return deserialize(
- response.getContent(),
- new TypeReference<SegmentIdWithShardSpec>()
- {
- }
- );
- }
- }
+ ) throws IOException;
- public void report(String supervisorTaskId, SubTaskReport report)
- {
- try {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "report",
- null,
- serialize(report),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "Failed to send taskReports to task[%s] with the HTTP code [%d]",
- supervisorTaskId,
- response.getStatus().getCode()
- );
- }
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
+ void report(SubTaskReport report);
Review Comment:
Sorry, I missed this comment prior to committing. I agree. However, the old method didn't have a javadoc either, so I don't think I made things worse.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #12696: Mid-level service client and updated high-level clients.
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #12696:
URL: https://github.com/apache/druid/pull/12696#discussion_r912896599
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskClient.java:
##########
@@ -19,122 +19,28 @@
package org.apache.druid.indexing.common.task.batch.parallel;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.druid.indexing.common.IndexTaskClient;
-import org.apache.druid.indexing.common.TaskInfoProvider;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.http.client.HttpClient;
-import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
-import org.jboss.netty.handler.codec.http.HttpMethod;
import org.joda.time.DateTime;
-import org.joda.time.Duration;
import javax.annotation.Nullable;
import java.io.IOException;
-public class ParallelIndexSupervisorTaskClient extends IndexTaskClient
+public interface ParallelIndexSupervisorTaskClient
{
- ParallelIndexSupervisorTaskClient(
- HttpClient httpClient,
- ObjectMapper objectMapper,
- TaskInfoProvider taskInfoProvider,
- Duration httpTimeout,
- String callerId,
- long numRetries
- )
- {
- super(httpClient, objectMapper, taskInfoProvider, httpTimeout, callerId, 1, numRetries);
- }
-
/**
* See {@link SinglePhaseParallelIndexTaskRunner#allocateNewSegment(String, DateTime)}.
*/
@Deprecated
- public SegmentIdWithShardSpec allocateSegment(String supervisorTaskId, DateTime timestamp) throws IOException
- {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "segment/allocate",
- null,
- serialize(timestamp),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "task[%s] failed to allocate a new segment identifier with the HTTP code[%d] and content[%s]",
- supervisorTaskId,
- response.getStatus().getCode(),
- response.getContent()
- );
- } else {
- return deserialize(
- response.getContent(),
- new TypeReference<SegmentIdWithShardSpec>()
- {
- }
- );
- }
- }
+ SegmentIdWithShardSpec allocateSegment(DateTime timestamp) throws IOException;
/**
* See {@link SinglePhaseParallelIndexTaskRunner#allocateNewSegment(String, DateTime, String, String)}.
*/
- public SegmentIdWithShardSpec allocateSegment(
- String supervisorTaskId,
+ SegmentIdWithShardSpec allocateSegment(
DateTime timestamp,
String sequenceName,
@Nullable String prevSegmentId
- ) throws IOException
- {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "segment/allocate",
- null,
- serialize(new SegmentAllocationRequest(timestamp, sequenceName, prevSegmentId)),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "task[%s] failed to allocate a new segment identifier with the HTTP code[%d] and content[%s]",
- supervisorTaskId,
- response.getStatus().getCode(),
- response.getContent()
- );
- } else {
- return deserialize(
- response.getContent(),
- new TypeReference<SegmentIdWithShardSpec>()
- {
- }
- );
- }
- }
+ ) throws IOException;
- public void report(String supervisorTaskId, SubTaskReport report)
- {
- try {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "report",
- null,
- serialize(report),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "Failed to send taskReports to task[%s] with the HTTP code [%d]",
- supervisorTaskId,
- response.getStatus().getCode()
- );
- }
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
+ void report(SubTaskReport report);
Review Comment:
Super Nit: as this is a interface, IMHO we should documentation to this method
##########
server/src/main/java/org/apache/druid/rpc/ServiceClientImpl.java:
##########
@@ -0,0 +1,424 @@
+/*
+ * 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.rpc;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.druid.java.util.common.Either;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
+import org.apache.druid.java.util.http.client.response.ObjectOrErrorResponseHandler;
+import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
+/**
+ * Production implementation of {@link ServiceClient}.
+ */
+public class ServiceClientImpl implements ServiceClient
Review Comment:
Should we implement this in an autoCloseable?
In case of a `close/shutdown` or an error condition
1. we should not schedule new async requests (infinite retry case)
2. Wait for existing requests to either timeout or abort them. I guess that would depend if the shutdown is gracefully or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on a diff in pull request #12696: Mid-level service client and updated high-level clients.
Posted by GitBox <gi...@apache.org>.
gianm commented on code in PR #12696:
URL: https://github.com/apache/druid/pull/12696#discussion_r908120965
##########
core/src/main/java/org/apache/druid/common/guava/FutureUtils.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.common.guava;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+
+public class FutureUtils
+{
+ /**
+ * Waits for a given future and returns its value, like {@code future.get()}.
+ *
+ * On InterruptedException, cancels the provided future if {@code cancelIfInterrupted}, then re-throws the
+ * original InterruptedException.
+ *
+ * Passes through CancellationExceptions and ExecutionExceptions as-is.
+ */
+ public static <T> T get(final ListenableFuture<T> future, final boolean cancelIfInterrupted)
+ throws InterruptedException, ExecutionException
+ {
+ try {
+ return future.get();
+ }
+ catch (InterruptedException e) {
+ if (cancelIfInterrupted) {
+ future.cancel(true);
+ }
+
Review Comment:
The InterruptedException is re-thrown here, so it's ok that we don't set the flag. According to https://docs.oracle.com/javase/tutorial/essential/concurrency/interrupt.html it is preferred to _not_ set the interrupt flag when throwing InterruptedException.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #12696: Mid-level service client and updated high-level clients.
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #12696:
URL: https://github.com/apache/druid/pull/12696#discussion_r912896599
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskClient.java:
##########
@@ -19,122 +19,28 @@
package org.apache.druid.indexing.common.task.batch.parallel;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.druid.indexing.common.IndexTaskClient;
-import org.apache.druid.indexing.common.TaskInfoProvider;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.http.client.HttpClient;
-import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
-import org.jboss.netty.handler.codec.http.HttpMethod;
import org.joda.time.DateTime;
-import org.joda.time.Duration;
import javax.annotation.Nullable;
import java.io.IOException;
-public class ParallelIndexSupervisorTaskClient extends IndexTaskClient
+public interface ParallelIndexSupervisorTaskClient
{
- ParallelIndexSupervisorTaskClient(
- HttpClient httpClient,
- ObjectMapper objectMapper,
- TaskInfoProvider taskInfoProvider,
- Duration httpTimeout,
- String callerId,
- long numRetries
- )
- {
- super(httpClient, objectMapper, taskInfoProvider, httpTimeout, callerId, 1, numRetries);
- }
-
/**
* See {@link SinglePhaseParallelIndexTaskRunner#allocateNewSegment(String, DateTime)}.
*/
@Deprecated
- public SegmentIdWithShardSpec allocateSegment(String supervisorTaskId, DateTime timestamp) throws IOException
- {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "segment/allocate",
- null,
- serialize(timestamp),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "task[%s] failed to allocate a new segment identifier with the HTTP code[%d] and content[%s]",
- supervisorTaskId,
- response.getStatus().getCode(),
- response.getContent()
- );
- } else {
- return deserialize(
- response.getContent(),
- new TypeReference<SegmentIdWithShardSpec>()
- {
- }
- );
- }
- }
+ SegmentIdWithShardSpec allocateSegment(DateTime timestamp) throws IOException;
/**
* See {@link SinglePhaseParallelIndexTaskRunner#allocateNewSegment(String, DateTime, String, String)}.
*/
- public SegmentIdWithShardSpec allocateSegment(
- String supervisorTaskId,
+ SegmentIdWithShardSpec allocateSegment(
DateTime timestamp,
String sequenceName,
@Nullable String prevSegmentId
- ) throws IOException
- {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "segment/allocate",
- null,
- serialize(new SegmentAllocationRequest(timestamp, sequenceName, prevSegmentId)),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "task[%s] failed to allocate a new segment identifier with the HTTP code[%d] and content[%s]",
- supervisorTaskId,
- response.getStatus().getCode(),
- response.getContent()
- );
- } else {
- return deserialize(
- response.getContent(),
- new TypeReference<SegmentIdWithShardSpec>()
- {
- }
- );
- }
- }
+ ) throws IOException;
- public void report(String supervisorTaskId, SubTaskReport report)
- {
- try {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "report",
- null,
- serialize(report),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "Failed to send taskReports to task[%s] with the HTTP code [%d]",
- supervisorTaskId,
- response.getStatus().getCode()
- );
- }
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
+ void report(SubTaskReport report);
Review Comment:
Super Nit: as this is a interface, IMHO we should add documentation to this method
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTaskClient.java:
##########
@@ -19,122 +19,28 @@
package org.apache.druid.indexing.common.task.batch.parallel;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.druid.indexing.common.IndexTaskClient;
-import org.apache.druid.indexing.common.TaskInfoProvider;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.http.client.HttpClient;
-import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
-import org.jboss.netty.handler.codec.http.HttpMethod;
import org.joda.time.DateTime;
-import org.joda.time.Duration;
import javax.annotation.Nullable;
import java.io.IOException;
-public class ParallelIndexSupervisorTaskClient extends IndexTaskClient
+public interface ParallelIndexSupervisorTaskClient
{
- ParallelIndexSupervisorTaskClient(
- HttpClient httpClient,
- ObjectMapper objectMapper,
- TaskInfoProvider taskInfoProvider,
- Duration httpTimeout,
- String callerId,
- long numRetries
- )
- {
- super(httpClient, objectMapper, taskInfoProvider, httpTimeout, callerId, 1, numRetries);
- }
-
/**
* See {@link SinglePhaseParallelIndexTaskRunner#allocateNewSegment(String, DateTime)}.
*/
@Deprecated
- public SegmentIdWithShardSpec allocateSegment(String supervisorTaskId, DateTime timestamp) throws IOException
- {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "segment/allocate",
- null,
- serialize(timestamp),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "task[%s] failed to allocate a new segment identifier with the HTTP code[%d] and content[%s]",
- supervisorTaskId,
- response.getStatus().getCode(),
- response.getContent()
- );
- } else {
- return deserialize(
- response.getContent(),
- new TypeReference<SegmentIdWithShardSpec>()
- {
- }
- );
- }
- }
+ SegmentIdWithShardSpec allocateSegment(DateTime timestamp) throws IOException;
/**
* See {@link SinglePhaseParallelIndexTaskRunner#allocateNewSegment(String, DateTime, String, String)}.
*/
- public SegmentIdWithShardSpec allocateSegment(
- String supervisorTaskId,
+ SegmentIdWithShardSpec allocateSegment(
DateTime timestamp,
String sequenceName,
@Nullable String prevSegmentId
- ) throws IOException
- {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "segment/allocate",
- null,
- serialize(new SegmentAllocationRequest(timestamp, sequenceName, prevSegmentId)),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "task[%s] failed to allocate a new segment identifier with the HTTP code[%d] and content[%s]",
- supervisorTaskId,
- response.getStatus().getCode(),
- response.getContent()
- );
- } else {
- return deserialize(
- response.getContent(),
- new TypeReference<SegmentIdWithShardSpec>()
- {
- }
- );
- }
- }
+ ) throws IOException;
- public void report(String supervisorTaskId, SubTaskReport report)
- {
- try {
- final StringFullResponseHolder response = submitSmileRequest(
- supervisorTaskId,
- HttpMethod.POST,
- "report",
- null,
- serialize(report),
- true
- );
- if (!isSuccess(response)) {
- throw new ISE(
- "Failed to send taskReports to task[%s] with the HTTP code [%d]",
- supervisorTaskId,
- response.getStatus().getCode()
- );
- }
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
+ void report(SubTaskReport report);
Review Comment:
Super Nit: as this is an interface, IMHO we should add documentation to this method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on a diff in pull request #12696: Mid-level service client and updated high-level clients.
Posted by GitBox <gi...@apache.org>.
gianm commented on code in PR #12696:
URL: https://github.com/apache/druid/pull/12696#discussion_r908121784
##########
core/src/main/java/org/apache/druid/common/guava/FutureUtils.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.common.guava;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+
+public class FutureUtils
+{
+ /**
+ * Waits for a given future and returns its value, like {@code future.get()}.
+ *
+ * On InterruptedException, cancels the provided future if {@code cancelIfInterrupted}, then re-throws the
+ * original InterruptedException.
+ *
+ * Passes through CancellationExceptions and ExecutionExceptions as-is.
+ */
+ public static <T> T get(final ListenableFuture<T> future, final boolean cancelIfInterrupted)
+ throws InterruptedException, ExecutionException
+ {
+ try {
+ return future.get();
+ }
+ catch (InterruptedException e) {
+ if (cancelIfInterrupted) {
+ future.cancel(true);
+ }
+
+ throw e;
+ }
+ }
+
+ /**
+ * Waits for a given future and returns its value, like {@code future.get()}.
+ *
+ * On InterruptException, cancels the provided future if {@code cancelIfInterrupted}, and in either case, throws
+ * a RuntimeException that wraps the original InterruptException.
+ *
+ * Passes through CancellationExceptions as-is.
+ *
+ * Re-wraps the causes of ExecutionExceptions using RuntimeException.
+ */
+ public static <T> T getUnchecked(final ListenableFuture<T> future, final boolean cancelIfInterrupted)
+ {
+ try {
+ return FutureUtils.get(future, cancelIfInterrupted);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
Review Comment:
Good call, I added a line to set the flag here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on a diff in pull request #12696: Mid-level service client and updated high-level clients.
Posted by GitBox <gi...@apache.org>.
gianm commented on code in PR #12696:
URL: https://github.com/apache/druid/pull/12696#discussion_r914000873
##########
server/src/main/java/org/apache/druid/rpc/ServiceClientImpl.java:
##########
@@ -0,0 +1,424 @@
+/*
+ * 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.rpc;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.druid.java.util.common.Either;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
+import org.apache.druid.java.util.http.client.response.ObjectOrErrorResponseHandler;
+import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
+/**
+ * Production implementation of {@link ServiceClient}.
+ */
+public class ServiceClientImpl implements ServiceClient
Review Comment:
Sorry, I missed this comment prior to committing. In the current design, the way you do this is to close the ServiceLocator. The ServiceLocator is marked Closeable, and once it's closed, any ServiceClients using it will stop doing retries and stop allowing new requests. The ServiceClient is meant to be stateless.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org