Add intermediary data server for shuffle #8088

jihoonson · 2019-07-16T01:42:51Z

First PR for #8061.

Description

This PR adds intermediary data server functionality to middleManagers. Documents for new configurations will be added once full functionalities for native parallel batch with shuffle is implemented.

This PR has:

been self-reviewed.
added documentation for new or modified features or behaviors.
added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
added unit tests or modified existing tests to cover new code paths.
been tested in a test Druid cluster.

…sever

himanshug

took a high level look today

himanshug · 2019-07-16T18:41:49Z

indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java

@@ -89,6 +94,13 @@ public TaskConfig(
    this.directoryLockTimeout = directoryLockTimeout == null
                                ? DEFAULT_DIRECTORY_LOCK_TIMEOUT
                                : directoryLockTimeout;
+    if (intermediarySegmentsLocations == null) {
+      this.intermediarySegmentsLocations = Collections.singletonList(
+          new StorageLocationConfig(new File(System.getProperty("java.io.tmpdir"), "intermediary-segments"), null, null)


Suggested change

new StorageLocationConfig(new File(System.getProperty("java.io.tmpdir"), "intermediary-segments"), null, null)

new StorageLocationConfig(new File(defaultDir(null, "intermediary-segments")), null, null)

himanshug · 2019-07-16T18:48:35Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/IntermediaryDataManager.java

+          }
+
+          if (exception != null) {
+            throw exception;


this code is running in a scheduler , why do we throw the overall exception in the end instead of logging them as they happened ?

Forgot to fix this before commit. Thanks.

can't we log the exceptions as they occur instead of accumulating all of them inside one and then logging it in the end ?

Ah sure, fixed.

himanshug · 2019-07-16T18:49:39Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/IntermediaryDataManager.java

+    if (supervisorTaskChecker != null) {
+      supervisorTaskChecker.shutdownNow();
+      supervisorTaskChecker.awaitTermination(10, TimeUnit.SECONDS);
+      supervisorTaskChecker = null;


what is the use of setting it null here ?

I think I did out of habit. Why?

it piqued my interest as usually you would do this sort of thing if you wanted to make something eligible for GC if it was a big object and part of large method so as to let it be GC'd asap or if you wanted to do something along the lines of making IntermediaryDataManager restartable etc.
I was trying to understand why you did that, but sounds like there is no reason so better to remove it :)

himanshug · 2019-07-16T18:53:05Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/IntermediaryDataManager.java

+   * Write a segment into one of configured locations. The location to write is chosen in a round-robin manner per
+   * supervisorTaskId.
+   *
+   * This method is only useful for the new Indexer model, and must not be called when tasks are running in the existing


does this mean parallel index task with shuffle support would not work with existing MMs?

why can't task processes have intermediary segment location propagated to them so that they could write data there and whole thing works for existing MMs as well ?

Shuffle system works with existing MMs. Tasks running in MMs just need to use another method instead of this one. Fixed the comment.

can you add what that other method is/would-be ? ... my understanding is that this method could be made static and peon process would write data by calling this method as well.

Add a new static method which could be used by MMs.

himanshug · 2019-07-16T18:54:59Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/http/ShuffleResource.java

+  @Path("/task/{supervisorTaskId}")
+  public Response deletePartitions(
+      @PathParam("supervisorTaskId") String supervisorTaskId,
+      @QueryParam("dataSource") String dataSource


It's.. used in TaskShuffleResourceFilter to check authorization. I know this is weird and, ideally, the resourceFilter should check authorization for task data instead of dataSource. However, the current security system only supports dataSource-level authorization. Added a comment.

ok... but why does it need to be declared here ? I understand that url needs that in query param for the filter to work but we probably don't need to capture that param in here.

Hmm, I just removed dataSource and changed to use StateResourceFilter instead. Added the below javadoc.

We use StateResourceFilter here because it performs an admin-like authorization and all endpoints here are supposed to be used for only internal communcation. Another possible alternate could be performing datasource-level authorization as in TaskResourceFilter. However, datasource information is not available in middleManagers or indexers yet which makes hard to use it. We could develop a new ResourceFileter in the future if needed.

clintropolis · 2019-07-17T00:58:08Z

...est/java/org/apache/druid/indexing/worker/IntermediaryDataManagerManualAddAndDeleteTest.java

+  }
+
+  @Test
+  public void testAddSegment() throws IOException


nit: is this test worth having since it is also implicitly tested by the other 2 tests?

Hmm, removed.

clintropolis · 2019-07-17T03:09:36Z

server/src/main/java/org/apache/druid/server/http/security/TaskShuffleResourceFilter.java

+ * This resource filter is used for data shuffle between native parallel index tasks. See ShuffleResource for details.
+ *
+ * It currently performs the authorization check for DATASOURCE, but ideally, it should be the authorization check
+ * for task data. This issue should be addressed in the future.


Could you clarify in these javadocs that I think you are referring to this should be doing what TaskResourceFilter is doing?

Removed this class now in favor of using StateResourceFilter.

himanshug · 2019-07-17T17:37:55Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/IntermediaryDataManager.java

+            }
+          }
+
+          final Map<String, TaskStatus> taskStatuses = indexingServiceClient.getTaskStatuses(expiredSupervisorTasks);


this has potential to throw exceptions and that would suppress further execution of this task... we should be stopping early only if the exception happened due to interrupt

usually it would be good to put scheduled task inside a big try-catch to avoid accidental premature exit or at least log what happened.

Sounds good. Added.

jihoonson · 2019-07-17T20:54:05Z

@himanshug and @clintropolis, thank you for taking a look. I addressed your comments.

I also noticed that the state of StorageLocation was not updated properly before. I refactored StorageLocation to be a bit more decoupled with DataSegment and fixed it to update its state properly.

clintropolis

overall lgtm 👍

clintropolis · 2019-07-17T22:51:42Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/http/ShuffleResource.java

+ * all endpoints here are supposed to be used for only internal communcation.
+ * Another possible alternate could be performing datasource-level authorization as in TaskResourceFilter.
+ * However, datasource information is not available in middleManagers or indexers yet which makes hard to use it.
+ * We could develop a new ResourceFileter in the future if needed.


clintropolis · 2019-07-17T23:38:31Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/IntermediaryDataManager.java

+    supervisorTaskChecker = Execs.scheduledSingleThreaded("intermediary-data-manager-%d");
+    // Discover partitions for new supervisorTasks
+    supervisorTaskChecker.scheduleAtFixedRate(
+        () -> {


This maybe seems large and complex enough to split into it's own method

Extracted as a method.

clintropolis · 2019-07-17T23:38:56Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/IntermediaryDataManager.java

+    // Note that the overlord sends a cleanup request when a supervisorTask is finished. The below check is to trigger
+    // the self-cleanup for when middleManager misses the cleanup request from the overlord.
+    supervisorTaskChecker.scheduleAtFixedRate(
+        () -> {


Same comment about splitting lambda into a method

Extracted as a method.

clintropolis · 2019-07-17T23:39:50Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/IntermediaryDataManager.java

+    throw new ISE("Can't find location to handle segment[%s]", segment);
+  }
+
+  public List<File> findPartitionFiles(String supervisorTaskId, Interval interval, int partitionId)


nit: could you put this and the non-static method that follows up with the other non-static methods?

Moved around to put non-static methods together.

clintropolis · 2019-07-17T23:44:25Z

CI failures look legitimate:

[ERROR] Failures: 
[ERROR] org.apache.druid.segment.loading.StorageLocationTest.testStorageLocation(org.apache.druid.segment.loading.StorageLocationTest)
[ERROR]   Run 1: StorageLocationTest.testStorageLocation:79->verifyLoc:99 expected:<990> but was:<980>
[ERROR]   Run 2: StorageLocationTest.testStorageLocation:79->verifyLoc:99 expected:<990> but was:<980>
[ERROR]   Run 3: StorageLocationTest.testStorageLocation:79->verifyLoc:99 expected:<990> but was:<980>
[ERROR]   Run 4: StorageLocationTest.testStorageLocation:79->verifyLoc:99 expected:<990> but was:<980>

jihoonson · 2019-07-18T00:05:31Z

@clintropolis thanks, fixed the test failure.

himanshug · 2019-07-18T17:11:53Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/IntermediaryDataManager.java

+          }
+          catch (Exception e) {
+            log.warn(e, "Error while cleaning up partitions for expired supervisors");
+          }


need to exit on "InterruptedException" or e.cause being InterruptedException

warn means no action needed, but in this case user might need to come in and do manual cleanup , so error level log is preferable.

actually it might be better for deleteExpiredSuprevisorTaskPartitionsIfNotRunning() to throw InterruptedException so that code here doesn't need to concern itself with e.cause check as that is internal detail of that method.

Oh yeah, forgot to handle that. Thanks. 👍

Just changed indexingServiceClient.getTaskStatuses to throw InterruptedException. I think it's better since it's a blocking call.

sorry, now that I see the change .. I realized it wasn't strictly necessary in this context ... but nice to have.

Yeah, I think it's better to log as an error rather than warning.

jihoonson · 2019-07-18T19:30:30Z

@himanshug thanks for the review!

jihoonson added 4 commits July 15, 2019 18:10

Add intermediary data server for shuffle

c29fd16

Merge branch 'master' of github.com:druid-io/druid into intermediary-…

f924030

…sever

javadoc

d25f0f2

adjust timeout

c4df069

jihoonson added the Area - Batch Ingestion label Jul 16, 2019

jihoonson added 3 commits July 15, 2019 19:42

resolved todo

a0aa7a8

fix test

c92105f

style

233f884

himanshug reviewed Jul 16, 2019

View reviewed changes

jihoonson added 2 commits July 16, 2019 13:33

address comments

d23788e

rename to shuffleDataLocations

402616c

clintropolis reviewed Jul 17, 2019

View reviewed changes

himanshug reviewed Jul 17, 2019

View reviewed changes

Address comments

f578196

bit adjustment StorageLocation

0929c01

clintropolis approved these changes Jul 17, 2019

View reviewed changes

jihoonson added 2 commits July 17, 2019 16:53

fix test

db03355

address comment & fix test

71a91ec

himanshug reviewed Jul 18, 2019

View reviewed changes

handle interrupted exception

cfc18b5

himanshug approved these changes Jul 18, 2019

View reviewed changes

jihoonson merged commit c7eb7cd into apache:master Jul 18, 2019

clintropolis added this to the 0.16.0 milestone Aug 8, 2019

	new StorageLocationConfig(new File(System.getProperty("java.io.tmpdir"), "intermediary-segments"), null, null)
	new StorageLocationConfig(new File(defaultDir(null, "intermediary-segments")), null, null)

Add intermediary data server for shuffle #8088

Add intermediary data server for shuffle #8088

Conversation

jihoonson commented Jul 16, 2019 • edited Loading

Description

himanshug left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

himanshug Jul 17, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jihoonson commented Jul 17, 2019

clintropolis left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

clintropolis commented Jul 17, 2019

jihoonson commented Jul 18, 2019

himanshug Jul 18, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jihoonson commented Jul 18, 2019

jihoonson commented Jul 16, 2019 •

edited

Loading

himanshug Jul 17, 2019 •

edited

Loading

himanshug Jul 18, 2019 •

edited

Loading