Supports Force Committing Segments in Batches #14811

noob-se7en · 2025-01-14T15:37:39Z

Problem Statement
The Force Commit API can cause high ingestion lag and slower queries because it triggers the simultaneous segment commit for all consuming segments. This happens because:

If N is the number of partition groups a server is consuming from, The API will cause all of the N consuming segments to commit. Hence N consumer threads will rush to acquire segment build semaphore. If the Semaphore allows only M permits, Only M consuming segments are in the segment build stage and the remaining (N - M) consumer threads are waiting on the Semaphore. Since (N - M) consumer threads are waiting, the consumption lag can become substantial.
Since M consuming segments are built in parallel, queries can become slower on poorly sized servers due to high memory consumption.

Solution
Adds an additional optional Integer parameter: batchSize to the forceCommit API (Default Value = Integer.MAX_VALUE i.e. commit as many segments as possible at once if no batchSize is provided) .

3 New query params added to the ForceCommit API:

batchSize (integer, optional):
Max number of consuming segments to commit at once.
Default: Integer.MAX_VALUE
Example: batchSize=100
batchStatusCheckIntervalSec (integer, optional):
How often (in seconds) to check whether the current batch of segments has been successfully committed.
Default: 5
Example: batchStatusCheckIntervalSec=10
batchStatusCheckTimeoutSec (integer, optional):
Timeout (in seconds) after which the controller stops checking the forceCommit status and throws an exception.
Default: 180
Example: batchStatusCheckTimeoutSec=300

codecov-commenter · 2025-01-14T16:51:46Z

Codecov Report

Attention: Patch coverage is 53.48837% with 40 lines in your changes missing coverage. Please review.

Project coverage is 63.68%. Comparing base (59551e4) to head (5ea7c3f).
Report is 1643 commits behind head on master.

Files with missing lines	Patch %	Lines
.../core/realtime/PinotLLCRealtimeSegmentManager.java	45.76%	28 Missing and 4 partials ⚠️
...ller/api/resources/PinotRealtimeTableResource.java	0.00%	6 Missing ⚠️
...ntroller/api/resources/ForceCommitBatchConfig.java	90.47%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #14811      +/-   ##
============================================
+ Coverage     61.75%   63.68%   +1.93%     
- Complexity      207     1471    +1264     
============================================
  Files          2436     2710     +274     
  Lines        133233   151974   +18741     
  Branches      20636    23468    +2832     
============================================
+ Hits          82274    96785   +14511     
- Misses        44911    47913    +3002     
- Partials       6048     7276    +1228

Flag	Coverage Δ
custom-integration1	`100.00% <ø> (+99.99%)`	⬆️
integration	`100.00% <ø> (+99.99%)`	⬆️
integration1	`100.00% <ø> (+99.99%)`	⬆️
integration2	`0.00% <ø> (ø)`
java-11	`63.67% <53.48%> (+1.96%)`	⬆️
java-21	`63.57% <53.48%> (+1.95%)`	⬆️
skip-bytebuffers-false	`63.68% <53.48%> (+1.93%)`	⬆️
skip-bytebuffers-true	`63.56% <53.48%> (+35.83%)`	⬆️
temurin	`63.68% <53.48%> (+1.93%)`	⬆️
unittests	`63.68% <53.48%> (+1.93%)`	⬆️
unittests1	`56.21% <ø> (+9.32%)`	⬆️
unittests2	`34.04% <53.48%> (+6.31%)`	⬆️

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

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Jackie-Jiang · 2025-01-15T03:45:01Z

This solves #11950

Jackie-Jiang

Seems you are pushing the batch throttling to the server side. What will happen if some replicas decide to commit, and others get throttled? Even worse, could this cause deadlock?

siddharthteotia · 2025-01-15T07:25:35Z

Is it not possible to solve the problem on controller / coordinate from the controller ? Pushing this down to the individual server will likely lead to error-prone situations

…ce_commit

noob-se7en · 2025-01-15T18:48:26Z

@Jackie-Jiang I don't quite get what is meant by

and others get throttled?

Regarding Deadlock or any edge case - Server will use the same logic which is used /tables/forceCommitStatus/{jobId} to check the status of the batch, so there should be no deadlock.

…mit_status

… into add_batch_force_commit

KKcorps · 2025-01-17T11:25:08Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

-    sendForceCommitMessageToServers(tableNameWithType, targetConsumingSegments);
+
+    List<Set<String>> segmentBatchList = getSegmentBatchList(idealState, targetConsumingSegments, batchSize);
+    ExecutorService executorService = Executors.newFixedThreadPool(1);


Do not create new thread pool on every request

ohh right!
Fixed (created a separate bounded ExecutorService for forceCommitAPI)

KKcorps · 2025-01-17T11:25:36Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+    sendForceCommitMessageToServers(tableNameWithType, segmentBatchToCommit);
+
+    try {
+      Thread.sleep(FORCE_COMMIT_STATUS_CHECK_INTERVAL_MS);


Why sleep at start?

This sleep is to save zookeeper calls, as at this point of code it's almost certain that all segments are yet to be committed.
(we need a new retryPolicy with initial start delay)

KKcorps · 2025-01-17T11:43:04Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+        String state = instanceToStateMap.get(instance);
+        if (state.equals(SegmentStateModel.CONSUMING)) {
+          instanceToConsumingSegments.putIfAbsent(instance, new LinkedList<>());
+          instanceToConsumingSegments.get(instance).add(segmentName);


You can simply this by using compute instead of putIfAbsent and get

KKcorps · 2025-01-17T11:45:42Z

...java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java

+    List<Set<String>> segmentBatchList =
+        realtimeSegmentManager.getSegmentBatchList(idealState, targetConsumingSegment, 2);
+
+    // i1 = [seg0, seg4, seg5, seg6]


remove these comments

KKcorps · 2025-01-17T11:47:25Z

...ller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java

@@ -170,19 +169,28 @@ public Map<String, String> forceCommit(
      @ApiParam(value = "Comma separated list of partition group IDs to be committed") @QueryParam("partitions")
      String partitionGroupIds,
      @ApiParam(value = "Comma separated list of consuming segments to be committed") @QueryParam("segments")
-      String consumingSegments, @Context HttpHeaders headers) {
+      String consumingSegments,
+      @ApiParam(value = "Max number of consuming segments to commit at once") @QueryParam("batchSize")


mention that default is all consuming segments at once in description

KKcorps · 2025-01-17T11:49:17Z

Please update PR description to reflect the change in API signature

…mit_status

… into add_batch_force_commit

…_commit

Jackie-Jiang

Mostly good

Jackie-Jiang · 2025-01-27T23:07:11Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

@@ -213,6 +222,7 @@ public PinotLLCRealtimeSegmentManager(PinotHelixResourceManager helixResourceMan
        controllerConf.getDeepStoreRetryUploadParallelism()) : null;
    _deepStoreUploadExecutorPendingSegments =
        _isDeepStoreLLCSegmentUploadRetryEnabled ? ConcurrentHashMap.newKeySet() : null;
+    _forceCommitExecutorService = Executors.newFixedThreadPool(4);


Having a fixed size pool could actually cause problems when there are multiple force commit request. Since it is waiting most of the time, I'd actually suggest making a single thread pool for each request same as the last version. It is not query path so the overhead should be fine.

Jackie-Jiang · 2025-01-27T23:10:11Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+
+    Map<String, Map<String, String>> segmentNameToInstanceToStateMap = idealState.getRecord().getMapFields();
+    for (String segmentName : segmentNameToInstanceToStateMap.keySet()) {
+      if (!targetConsumingSegments.contains(segmentName)) {


Let's loop over targetConsumingSegments instead of ideal state. Ideal state should always contain targetConsumingSegments because they are extracted from ideal state.

Jackie-Jiang · 2025-01-27T23:13:56Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+          instanceToConsumingSegments.compute(instance, (key, value) -> {
+            if (value == null) {
+              value = new LinkedList<>();
+            }
+            value.add(segmentName);
+            return value;
+          });


Suggested change

instanceToConsumingSegments.compute(instance, (key, value) -> {

if (value == null) {

value = new LinkedList<>();

}

value.add(segmentName);

return value;

});

instanceToConsumingSegments.computeIfAbsent(instance, k -> new LinkedList<>()).add(segmentName);

Jackie-Jiang · 2025-01-27T23:15:18Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+      for (String instance : instanceToStateMap.keySet()) {
+        String state = instanceToStateMap.get(instance);


Use entrySet() to reduce lookup

Jackie-Jiang · 2025-01-27T23:17:27Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+
+    while (segmentsRemaining) {
+      segmentsRemaining = false;
+      // pick segments in round-robin fashion to parallelize


Jackie-Jiang · 2025-01-27T23:18:59Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+          String segmentName = queue.poll();
+          // there might be a segment replica hosted on
+          // another instance added before
+          if (segmentsAdded.contains(segmentName)) {


We can reduce a lookup by

Suggested change

if (segmentsAdded.contains(segmentName)) {

if (!segmentsAdded.add(segmentName)) {

Jackie-Jiang · 2025-01-27T23:22:29Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+      // pick segments in round-robin fashion to parallelize
+      // forceCommit across max servers
+      for (Queue<String> queue : instanceToConsumingSegments.values()) {
+        if (!queue.isEmpty()) {


We can remove the queue when it is empty to avoid checking it again and again. You may use iterator to remove entry without extra lookup

Jackie-Jiang · 2025-01-27T23:26:22Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+
+    try {
+      Thread.sleep(FORCE_COMMIT_STATUS_CHECK_INTERVAL_MS);
+    } catch (InterruptedException ignored) {


Ignoring interrupt could be risky (holding a long running thread). Let's wrap it as a RuntimeException and throw it. We may log an error when catching it

Jackie-Jiang · 2025-01-27T23:27:30Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

+    }
+
+    int attemptCount = 0;
+    final Set<String>[] segmentsYetToBeCommitted = new Set[]{new HashSet<>()};


Suggested change

final Set<String>[] segmentsYetToBeCommitted = new Set[]{new HashSet<>()};

final Set<String>[] segmentsYetToBeCommitted = new Set[1];

Jackie-Jiang · 2025-01-27T23:33:39Z

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

@@ -152,6 +157,9 @@ public class PinotLLCRealtimeSegmentManager {

  // Max time to wait for all LLC segments to complete committing their metadata while stopping the controller.
  private static final long MAX_LLC_SEGMENT_METADATA_COMMIT_TIME_MILLIS = 30_000L;
+  private static final int FORCE_COMMIT_STATUS_CHECK_INTERVAL_MS = 15000;


Let's take the check interval also from the rest API because different use case might want different interval; we might also want to add a TIMEOUT and also take that from rest API. The retry count can be calculated from timeout and interval.
We can provide default values (e.g. 5s, 3min) for them in case they are not provided. IMO 15s interval is too long because it means for each batch we will wait at least 15s.

sure.
(Default case (5s interval, 3m timeout) might be little expensive as we are waiting for segments build to complete and there will be 36 calls at max to ZK per batch)

noob-se7en added 13 commits January 13, 2025 01:07

Supports batching in ForceCommit API

eeb5be1

nit

5f5a554

Refactoring

ca5104a

nit

434e8a3

nit

504f3c9

nit

987bb00

nit

ff25c5f

nit

e28ff47

nit

99a7cee

lint

255bc34

nit

3a9e41a

fixes lint

b2eeb85

nit

1782207

Jackie-Jiang added feature ingestion rest-api labels Jan 15, 2025

Jackie-Jiang reviewed Jan 15, 2025

View reviewed changes

noob-se7en added 8 commits January 15, 2025 21:43

Merge branch 'master' of github.com:Harnoor7/pinot into add_batch_for…

90db3b8

…ce_commit

refactoring

fa418b9

refactoring

470c6eb

fixes bug

8de7bfc

nit

4f2d4fc

nit

50af02e

nit

09d557e

nit

32b7fd5

noob-se7en requested a review from Jackie-Jiang January 15, 2025 18:49

noob-se7en added 4 commits January 17, 2025 16:32

Merge branch 'master' of github.com:apache/pinot into update_forceCom…

b8a2e7f

…mit_status

addresses PR comments

5730a06

nit

c8565d6

Merge branch 'update_forceCommit_status' of github.com:Harnoor7/pinot…

165e7ab

… into add_batch_force_commit

KKcorps reviewed Jan 17, 2025

View reviewed changes

noob-se7en added 3 commits January 18, 2025 00:51

refactoring

0cab772

Addresses PR comments

de04824

nit

b95a2f6

noob-se7en requested a review from KKcorps January 17, 2025 20:14

noob-se7en added 8 commits January 23, 2025 15:59

attempt to fix test

857dd6a

Merge branch 'master' of github.com:apache/pinot into update_forceCom…

2f7e5d9

…mit_status

nit

0b64439

Attempts to fix test

9e3ddad

Merge branch 'update_forceCommit_status' of github.com:Harnoor7/pinot…

01604e9

… into add_batch_force_commit

Merge branch 'master' of github.com:apache/pinot into add_batch_force…

bb84ae2

…_commit

Attempts to fix test

71f4ee1

attempt to fix test

2408d13

Jackie-Jiang reviewed Jan 27, 2025

View reviewed changes

noob-se7en added 6 commits January 29, 2025 00:24

Addresses PR comments

ff67929

Adds timeout and interval query parameters in API

80dda07

nit

11299f4

fixes lint

ad7aec0

Adds unit test

7ea5535

nit

5ea7c3f

noob-se7en requested a review from Jackie-Jiang January 29, 2025 09:35

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Supports Force Committing Segments in Batches #14811

Supports Force Committing Segments in Batches #14811

noob-se7en commented Jan 14, 2025 •

edited

Loading

codecov-commenter commented Jan 14, 2025 •

edited

Loading

Jackie-Jiang commented Jan 15, 2025

Jackie-Jiang left a comment

siddharthteotia commented Jan 15, 2025

noob-se7en commented Jan 15, 2025

KKcorps Jan 17, 2025

noob-se7en Jan 17, 2025

KKcorps Jan 17, 2025

noob-se7en Jan 17, 2025

KKcorps Jan 17, 2025 •

edited

Loading

KKcorps Jan 17, 2025

KKcorps Jan 17, 2025

KKcorps commented Jan 17, 2025 •

edited

Loading

Jackie-Jiang left a comment

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

Jackie-Jiang Jan 27, 2025

noob-se7en Jan 28, 2025

		for (String instance : instanceToStateMap.keySet()) {
		String state = instanceToStateMap.get(instance);

	if (segmentsAdded.contains(segmentName)) {
	if (!segmentsAdded.add(segmentName)) {

	final Set<String>[] segmentsYetToBeCommitted = new Set[]{new HashSet<>()};
	final Set<String>[] segmentsYetToBeCommitted = new Set[1];

Supports Force Committing Segments in Batches #14811

Are you sure you want to change the base?

Supports Force Committing Segments in Batches #14811

Conversation

noob-se7en commented Jan 14, 2025 • edited Loading

codecov-commenter commented Jan 14, 2025 • edited Loading

Codecov Report

Jackie-Jiang commented Jan 15, 2025

Jackie-Jiang left a comment

Choose a reason for hiding this comment

siddharthteotia commented Jan 15, 2025

noob-se7en commented Jan 15, 2025

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

KKcorps Jan 17, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KKcorps commented Jan 17, 2025 • edited Loading

Jackie-Jiang 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

noob-se7en commented Jan 14, 2025 •

edited

Loading

codecov-commenter commented Jan 14, 2025 •

edited

Loading

KKcorps Jan 17, 2025 •

edited

Loading

KKcorps commented Jan 17, 2025 •

edited

Loading