gianm commented on code in PR #16462:
URL: https://github.com/apache/druid/pull/16462#discussion_r1605668325
##########
server/src/main/java/org/apache/druid/rpc/indexing/SpecificTaskServiceLocator.java:
##########
@@ -209,18 +173,103 @@ public void close()
}
}
- private class TaskLocationFetcher
+ private void resolvePendingFuture(TaskState state, TaskLocation location)
{
- TaskLocation getLocation()
- {
- final TaskStatusResponse statusResponse = FutureUtils.getUnchecked(
- overlordClient.taskStatus(taskId),
- true
- );
- if (statusResponse == null || statusResponse.getStatus() == null) {
- return TaskLocation.unknown();
- } else {
- return statusResponse.getStatus().getLocation();
+ synchronized (lock) {
+ if (pendingFuture != null) {
+ lastKnownState = state;
+ lastKnownLocation = location == null ? null : new ServiceLocation(
+ location.getHost(),
+ location.getPort(),
+ location.getTlsPort(),
+ StringUtils.format("%s/%s", BASE_PATH,
StringUtils.urlEncode(taskId))
+ );
+
+ if (lastKnownState != TaskState.RUNNING) {
+ pendingFuture.set(ServiceLocations.closed());
+ } else if (lastKnownLocation == null) {
+
pendingFuture.set(ServiceLocations.forLocations(Collections.emptySet()));
+ } else {
+ pendingFuture.set(ServiceLocations.forLocation(lastKnownLocation));
+ }
+
+ // Clear pendingFuture once it has been set.
+ pendingFuture = null;
+ }
+ }
+ }
+
+ private void resolvePendingFutureOnException(Throwable t)
+ {
+ synchronized (lock) {
+ if (pendingFuture != null) {
+ pendingFuture.setException(t);
+
+ // Clear pendingFuture once it has been set.
+ pendingFuture = null;
+ }
+ }
+ }
+
+ /**
+ * Invokes the single task status API {@link OverlordClient#taskStatus} if
the
+ * multi-task status API returns an unknown location (this can happen if the
+ * Overlord is running on a version older than Druid 30.0.0 (pre #15724)).
+ */
+ private void fetchFallbackTaskLocation()
+ {
+ synchronized (lock) {
+ if (pendingFuture != null) {
+ final ListenableFuture<TaskStatusResponse> taskStatusFuture;
+ try {
+ taskStatusFuture = overlordClient.taskStatus(taskId);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
Review Comment:
Should resolve `pendingFuture` in this case, I think. It would end up
abandoned if an exception is actually thrown 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
