[7338] Allow Reloading Segments with Multiple Threads #7893

suddendust · 2021-12-12T11:57:36Z

Description

This change allows users to reload segments of a table in parallel. More description here.

Upgrade Notes

Does this PR prevent a zero down-time upgrade? (Assume upgrade order: Controller, Broker, Server, Minion)

Yes

Does this PR fix a zero-downtime upgrade introduced earlier?

Yes

Does this PR otherwise need attention when creating release notes? Things to consider:

Yes

… API

suddendust · 2021-12-12T12:11:55Z

Working on tests for this.

codecov-commenter · 2021-12-12T12:52:18Z

Codecov Report

Merging #7893 (b486e5a) into master (fb12a50) will decrease coverage by 0.02%.
The diff coverage is 76.31%.

@@             Coverage Diff              @@
##             master    #7893      +/-   ##
============================================
- Coverage     71.34%   71.31%   -0.03%     
- Complexity     4087     4194     +107     
============================================
  Files          1587     1595       +8     
  Lines         82071    82568     +497     
  Branches      12267    12320      +53     
============================================
+ Hits          58553    58885     +332     
- Misses        19550    19708     +158     
- Partials       3968     3975       +7

Flag	Coverage Δ
integration1	`29.02% <76.31%> (-0.26%)`	⬇️
integration2	`27.66% <71.05%> (+0.01%)`	⬆️
unittests1	`68.23% <0.00%> (-0.10%)`	⬇️
unittests2	`14.29% <0.00%> (-0.15%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...server/starter/helix/HelixInstanceDataManager.java	`81.50% <69.23%> (+1.02%)`	⬆️
...pache/pinot/core/util/SegmentRefreshSemaphore.java	`72.22% <72.22%> (ø)`
...er/starter/helix/SegmentMessageHandlerFactory.java	`73.68% <100.00%> (+1.08%)`	⬆️
...data/manager/realtime/DefaultSegmentCommitter.java	`0.00% <0.00%> (-80.00%)`	⬇️
...ry/optimizer/statement/JsonStatementOptimizer.java	`0.00% <0.00%> (-77.64%)`	⬇️
.../pinot/spi/exception/BadQueryRequestException.java	`66.66% <0.00%> (-33.34%)`	⬇️
...a/manager/realtime/RealtimeSegmentDataManager.java	`50.00% <0.00%> (-25.00%)`	⬇️
...in/stream/kafka20/KafkaPartitionLevelConsumer.java	`66.66% <0.00%> (-20.00%)`	⬇️
...er/api/resources/LLCSegmentCompletionHandlers.java	`43.56% <0.00%> (-18.82%)`	⬇️
...data/manager/realtime/SegmentCommitterFactory.java	`88.23% <0.00%> (-11.77%)`	⬇️
... and 95 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update fb12a50...b486e5a. Read the comment docs.

… parallelism

richardstartin · 2021-12-13T13:34:41Z

pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java

+      CompletableFuture.runAsync(() -> {
+        try {
+          reloadSegment(tableNameWithType, segmentMetadata, tableConfig, schema, forceDownload);
+        } catch (Exception e) {
+          String segmentName = segmentMetadata.getName();
+          LOGGER.error("Caught exception while reloading segment: {} in table: {}", segmentName, tableNameWithType, e);
+          failedSegments.add(segmentName);
+          sampleException.set(e);
+        } finally {
+          latch.countDown();
+        }
+      }, workers);


I think you can make better use of the CompletableFuture API here:

CompletableFuture.allOf(segmentsMetadata.stream() .map(segmentMetadata -> CompletableFuture.runAsync(() -> {... reload the segment ...}, workers) .toArray(CompletableFuture[]::new)) .get(timeout, TimeUnit.MILLISECONDS);

Which would eliminate the latch and make the code easier to reason about.

This is excellent. Thanks! Definitely need to revisit CompletableFuture's API once.

siddharthteotia · 2021-12-14T00:54:34Z

pinot-common/src/main/java/org/apache/pinot/common/messages/SegmentReloadMessage.java

@@ -49,6 +50,13 @@ public SegmentReloadMessage(@Nonnull String tableNameWithType, @Nullable String
    znRecord.setBooleanField(FORCE_DOWNLOAD_KEY, forceDownload);
  }

+  public SegmentReloadMessage(@Nonnull String tableNameWithType, @Nullable String segmentName, boolean forceDownload,
+      int parallelism) {


I know this is draft PR

Just curious on the approach. How does the user specify parallelism parameter ? Can we use semaphore like we do in refresh (which is also delivered to server via helix message) ?

@suddendust if this is a draft PR, can you please add WIP in the title? thanks

@siddharthteotia After some discussion, the conclusion is that there is no point in having yet another param to control the parallelism. Instead, using the existing _refreshThreadsSemaphore for this makes sense as reload and refresh are so similar. For each available permit, we reload a segment asynchronously.

Yes, sound good

suddendust · 2021-12-22T07:07:54Z

@Jackie-Jiang Requesting review, thanks!

Jackie-Jiang · 2021-12-22T23:09:55Z

pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java

      try {
+        acquireSema("ALL", refreshThreadSemaphore);


Put the segment name

Suggested change

acquireSema("ALL", refreshThreadSemaphore);

acquireSema(segmentsMetadata.getName(), refreshThreadSemaphore);

Jackie-Jiang · 2021-12-22T23:10:49Z

pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java

      try {
+        acquireSema("ALL", refreshThreadSemaphore);


acquireSema() should be outside of the try block

Yes, that was a miss. Thanks!

Jackie-Jiang · 2021-12-22T23:13:46Z

pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java

      }
-    }
+    }, workers)).toArray(CompletableFuture[]::new)).get();


Do you need to shut down the executor service after it is done?

It should be GCed after ~ 60s (TTL of cached threads). However, it's a good idea to shut it down explicitly. Addressed.

Jackie-Jiang · 2021-12-22T23:20:00Z

pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java

    LOGGER.info("Reloaded all segments in table: {}", tableNameWithType);
  }

+  private void acquireSema(String context, Semaphore refreshThreadSemaphore)


The first argument is segmentName

(optional) This part is duplicated. One way to avoid duplicate code is to add a SegmentRefreshSemaphore class to wrap this logic.

Created a wrapper class.

Jackie-Jiang

LGTM with minor comments

Jackie-Jiang · 2021-12-23T23:23:10Z

pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java

+        } catch (Exception e) {
+          LOGGER.error("Caught exception while reloading segment: {} in table: {}", segmentName, tableNameWithType, e);
+          failedSegments.add(segmentName);
+          sampleException.set(e);


This catch is duplicated. We only need one try/catch/finally here

The outside try-catch is to handle any InterruptedException while trying to acquire a permit (acquireSema throws an InterruptedException). If we put a finally in that block it'll try to release a permit in this case which would be wrong. So we need to another try-catch inside the outer try to execute the finally.

I see. In that case we should remove this catch block (keep the try/finally), and let the outside catch to handle the exception

Yes, that one is redundant. Addressed.

pinot-core/src/main/java/org/apache/pinot/core/util/SegmentRefreshSemaphore.java

…reshSemaphore.java Co-authored-by: Xiaotian (Jackie) Jiang <[email protected]>

suddendust added 5 commits December 12, 2021 14:17

Added reload parallelism to SegmentReloadMessage

5a626b7

Undo unwanted change

3be37c7

Accept parallelism param in PinotSegmentRestletResource

801b2b6

Added default parallelism value in PinotSegmentRestletResource reload…

bcc17b7

… API

Minor refactoring

3817fce

suddendust marked this pull request as draft December 12, 2021 11:57

Fixed checkstyle violations

f5f9eb8

suddendust mentioned this pull request Dec 12, 2021

Allow reloading segments with multiple threads #7338

Closed

Added @QueryParam in API signatures

98d1fe9

suddendust added 2 commits December 13, 2021 18:46

fix worker pool size

00047d8

added preconditions on parallelism param, added log when noSegments <…

a7602c1

… parallelism

richardstartin reviewed Dec 13, 2021

View reviewed changes

suddendust added 2 commits December 13, 2021 23:42

better use of completablefuture API as recommended

0d30a5b

spotless

5242862

siddharthteotia reviewed Dec 14, 2021

View reviewed changes

suddendust changed the title ~~[7338] Allow Reloading Segments with Multiple Threads~~ [7338] [WIP] Allow Reloading Segments with Multiple Threads Dec 15, 2021

suddendust added 3 commits December 21, 2021 22:55

Use existing _refreshThreadsSemaphore to control reload parallelism

7963329

Reverted unwanted changes

416b7e8

Removed signature change for InstanceDataManager#reloadSegment

14f608e

suddendust marked this pull request as ready for review December 21, 2021 19:16

suddendust changed the title ~~[7338] [WIP] Allow Reloading Segments with Multiple Threads~~ [7338] Allow Reloading Segments with Multiple Threads Dec 21, 2021

suddendust added 3 commits December 22, 2021 12:22

Nullcheck during sema release, minor refactoring

00f6a8d

fix checkstyle

6cc9635

refactor

80ebb4d

Jackie-Jiang reviewed Dec 22, 2021

View reviewed changes

suddendust added 2 commits December 23, 2021 16:11

Addressed comments

49ea862

Refactoring

db376e9

suddendust requested a review from Jackie-Jiang December 23, 2021 10:48

suddendust added 2 commits December 23, 2021 16:26

spotless

efb8a3f

Added missing header

ea7800b

Jackie-Jiang approved these changes Dec 23, 2021

View reviewed changes

suddendust and others added 3 commits December 24, 2021 11:02

Update pinot-core/src/main/java/org/apache/pinot/core/util/SegmentRef…

29c52d4

…reshSemaphore.java Co-authored-by: Xiaotian (Jackie) Jiang <[email protected]>

fixed compilation issue due to upstream commit

2fdb3c2

removed redundant catch block

b486e5a

Jackie-Jiang merged commit 99762cb into apache:master Dec 25, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[7338] Allow Reloading Segments with Multiple Threads #7893

[7338] Allow Reloading Segments with Multiple Threads #7893

suddendust commented Dec 12, 2021

suddendust commented Dec 12, 2021

codecov-commenter commented Dec 12, 2021 •

edited

Loading

richardstartin Dec 13, 2021

suddendust Dec 13, 2021

siddharthteotia Dec 14, 2021

mcvsubbu Dec 15, 2021

suddendust Dec 21, 2021

siddharthteotia Dec 27, 2021

suddendust commented Dec 22, 2021

Jackie-Jiang Dec 22, 2021

Jackie-Jiang Dec 22, 2021

suddendust Dec 23, 2021

Jackie-Jiang Dec 22, 2021

suddendust Dec 23, 2021 •

edited

Loading

Jackie-Jiang Dec 22, 2021

suddendust Dec 23, 2021

Jackie-Jiang left a comment

Jackie-Jiang Dec 23, 2021

suddendust Dec 24, 2021 •

edited

Loading

Jackie-Jiang Dec 24, 2021 •

edited

Loading

suddendust Dec 24, 2021

	acquireSema("ALL", refreshThreadSemaphore);
	acquireSema(segmentsMetadata.getName(), refreshThreadSemaphore);

[7338] Allow Reloading Segments with Multiple Threads #7893

[7338] Allow Reloading Segments with Multiple Threads #7893

Conversation

suddendust commented Dec 12, 2021

Description

Upgrade Notes

suddendust commented Dec 12, 2021

codecov-commenter commented Dec 12, 2021 • edited Loading

Codecov Report

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

suddendust commented Dec 22, 2021

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

suddendust Dec 23, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

suddendust Dec 24, 2021 • edited Loading

Choose a reason for hiding this comment

Jackie-Jiang Dec 24, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Dec 12, 2021 •

edited

Loading

suddendust Dec 23, 2021 •

edited

Loading

suddendust Dec 24, 2021 •

edited

Loading

Jackie-Jiang Dec 24, 2021 •

edited

Loading