You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2022/06/28 23:14:24 UTC
[GitHub] [pinot] sajjad-moradi opened a new pull request, #8986: Support pause/resume consumption of realtime tables
sajjad-moradi opened a new pull request, #8986:
URL: https://github.com/apache/pinot/pull/8986
## Description
This PR adds support for pause/resume consumption on realtime table without disabling the table. The following endpoints on controller can be used to pause/resume consumption:
- /tables/{tableName}/pauseConsumption
- /tables/{tableName}/resumeConsumption
- /tables/{tableName}/consumptionStatus
For more design details, please refer to this [doc](https://docs.google.com/document/d/1uuXuif0SfMTnH3ykfuzXXNjh-LyszjilodD_mdlilZg/edit#heading=h.reqaryi1pqib).
Note that this PR will solve the following issues:
[#6302](https://github.com/apache/pinot/issues/6302), [#7280](https://github.com/apache/pinot/issues/7280), [#6854](https://github.com/apache/pinot/issues/6854), [#6679](https://github.com/apache/pinot/issues/6679), [#7100](https://github.com/apache/pinot/issues/7100), and [#7039](https://github.com/apache/pinot/issues/7039).
## Testing Done
Verified the desired behavior locally by running LLRealtimeClusterIntegrationTest and using the endpoints to pause, resume, and get the consumption status.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] sajjad-moradi commented on a diff in pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on code in PR #8986:
URL: https://github.com/apache/pinot/pull/8986#discussion_r918204110
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java:
##########
@@ -169,6 +173,34 @@ public HelixTaskResult handleMessage()
}
}
+ private class ForceCommitMessageHandler extends DefaultMessageHandler {
+
+ private String _tableName;
+ private Set<String> _segmentNames;
+
+ public ForceCommitMessageHandler(ForceCommitMessage forceCommitMessage, ServerMetrics metrics,
+ NotificationContext ctx) {
+ super(forceCommitMessage, metrics, ctx);
+ _tableName = forceCommitMessage.getTableName();
+ _segmentNames = forceCommitMessage.getSegmentNames();
+ }
+
+ @Override
+ public HelixTaskResult handleMessage()
+ throws InterruptedException {
+ HelixTaskResult helixTaskResult = new HelixTaskResult();
+ _logger.info("Handling force commit message for table {} segments {}", _tableName, _segmentNames);
+ try {
+ _instanceDataManager.forceCommit(_tableName, _segmentNames);
+ helixTaskResult.setSuccess(true);
+ } catch (Exception e) {
+ _metrics.addMeteredTableValue(_tableNameWithType, ServerMeter.DELETE_TABLE_FAILURES, 1);
+ Utils.rethrowException(e);
Review Comment:
Helix doesn't retry. It just logs the exception.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1399,4 +1417,120 @@ public void uploadToDeepStoreIfMissing(TableConfig tableConfig, List<SegmentZKMe
}
}
}
+
+ /**
+ * Pause consumption on a table by
+ * 1) setting "isTablePaused" in ideal states to true and
+ * 2) sending force commit messages to servers
+ */
+ public PauseStatus pauseConsumption(String tableNameWithType) {
+ IdealState updatedIdealState = updatePauseStatusInIdealState(tableNameWithType, true);
+ Set<String> consumingSegments = findConsumingSegments(updatedIdealState);
+ sendForceCommitMessageToServers(tableNameWithType, consumingSegments);
+ return new PauseStatus(true, consumingSegments, consumingSegments.isEmpty() ? null : "Pause flag is set."
+ + " Consuming segments are being committed."
+ + " Use /pauseStatus endpoint in a few moments to check if all consuming segments have been committed.");
+ }
+
+ /**
+ * Resume consumption on a table by
+ * 1) setting "isTablePaused" in ideal states to false and
+ * 2) triggering segment validation job to create new consuming segments in ideal states
+ */
+ public PauseStatus resumeConsumption(String tableNameWithType) {
+ IdealState updatedIdealState = updatePauseStatusInIdealState(tableNameWithType, false);
+
+ // trigger realtime segment validation job to resume consumption
+ Map<String, String> taskProperties = new HashMap<>();
+ taskProperties.put(RealtimeSegmentValidationManager.RECREATE_DELETED_CONSUMING_SEGMENT_KEY, "true");
Review Comment:
Created the issue: #9039
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1399,4 +1417,120 @@ public void uploadToDeepStoreIfMissing(TableConfig tableConfig, List<SegmentZKMe
}
}
}
+
+ /**
+ * Pause consumption on a table by
+ * 1) setting "isTablePaused" in ideal states to true and
+ * 2) sending force commit messages to servers
+ */
+ public PauseStatus pauseConsumption(String tableNameWithType) {
+ IdealState updatedIdealState = updatePauseStatusInIdealState(tableNameWithType, true);
+ Set<String> consumingSegments = findConsumingSegments(updatedIdealState);
+ sendForceCommitMessageToServers(tableNameWithType, consumingSegments);
+ return new PauseStatus(true, consumingSegments, consumingSegments.isEmpty() ? null : "Pause flag is set."
Review Comment:
This is not internally used. We return the pause status to the client.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] sajjad-moradi commented on pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on PR #8986:
URL: https://github.com/apache/pinot/pull/8986#issuecomment-1229656974
> @sajjad-moradi would you please add this to the documentation? (or point me to it if you already have, I couldn't find it easily). I think this would be a good place: https://docs.pinot.apache.org/basics/data-import/pinot-stream-ingestion, for all of them pause/resume/forceCommit/change stream.
I'll try to update the doc this coming week.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] sajjad-moradi commented on pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on PR #8986:
URL: https://github.com/apache/pinot/pull/8986#issuecomment-1170278451
> You should also check the idealstate once before entering the consumeLoop()
That would be for handling a rare edge case. The probability of that case happening is very low and can be calculated as `t1/t2` where
`t1`: time duration between receiving online-to-consuming msg and entering the consumeLoop
`t2`: consumeLoop duration
Note that t1 is in microseconds-milliseconds and t2 is in hours (depending on table config).
If that case happens, the partition will eventually be paused when after the segment gets completed. The operator can reissue the pause request if they need it sooner.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] sajjad-moradi commented on a diff in pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on code in PR #8986:
URL: https://github.com/apache/pinot/pull/8986#discussion_r918209300
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -49,27 +50,71 @@
HttpHeaders.AUTHORIZATION, in = ApiKeyAuthDefinition.ApiKeyLocation.HEADER, key = SWAGGER_AUTHORIZATION_KEY)))
@Path("/")
public class PinotRealtimeTableResource {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PinotRealtimeTableResource.class);
+
@Inject
PinotHelixResourceManager _pinotHelixResourceManager;
+ @Inject
+ PinotLLCRealtimeSegmentManager _pinotLLCRealtimeSegmentManager;
+
+ @POST
+ @Path("/tables/{tableName}/pauseConsumption")
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Pause consumption of a realtime table",
+ notes = "Pause the consumption of a realtime table")
+ public Response pauseConsumption(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName) {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ validate(tableNameWithType);
+ try {
+ return Response.ok(_pinotLLCRealtimeSegmentManager.pauseConsumption(tableNameWithType)).build();
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
+
@POST
@Path("/tables/{tableName}/resumeConsumption")
@Produces(MediaType.APPLICATION_JSON)
- @Consumes(MediaType.APPLICATION_JSON)
- @ApiOperation(value = "Resume the consumption of a realtime table",
- notes = "Resume the consumption of a realtime table")
- public String resumeConsumption(
- @ApiParam(value = "Name of the table", required = true)
- @PathParam("tableName") String tableName) throws JsonProcessingException {
- // TODO: Add util method for invoking periodic tasks
+ @ApiOperation(value = "Resume consumption of a realtime table",
+ notes = "Resume the consumption for a realtime table")
+ public Response resumeConsumption(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName) {
String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
- Map<String, String> taskProperties = new HashMap<>();
- taskProperties.put(RealtimeSegmentValidationManager.RECREATE_DELETED_CONSUMING_SEGMENT_KEY, "true");
+ validate(tableNameWithType);
+ try {
+ return Response.ok(_pinotLLCRealtimeSegmentManager.resumeConsumption(tableNameWithType)).build();
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
- Pair<String, Integer> taskExecutionDetails = _pinotHelixResourceManager
- .invokeControllerPeriodicTask(tableNameWithType, Constants.REALTIME_SEGMENT_VALIDATION_MANAGER, taskProperties);
+ @GET
+ @Path("/tables/{tableName}/pauseStatus")
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Return pause status of a realtime table",
+ notes = "Return pause status of a realtime table along with list of consuming segments.")
+ public Response getConsumptionStatus(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName) {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ validate(tableNameWithType);
+ try {
+ return Response.ok().entity(_pinotLLCRealtimeSegmentManager.getPauseStatus(tableNameWithType)).build();
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
- return "{\"Log Request Id\": \"" + taskExecutionDetails.getLeft()
- + "\",\"Controllers notified\":" + (taskExecutionDetails.getRight() > 0) + "}";
+ private void validate(String tableNameWithType) {
+ IdealState idealState = _pinotHelixResourceManager.getTableIdealState(tableNameWithType);
+ if (idealState == null) {
+ throw new ControllerApplicationException(LOGGER, "Ideal State is null for table " + tableNameWithType,
Review Comment:
Done.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] sajjad-moradi commented on a diff in pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on code in PR #8986:
URL: https://github.com/apache/pinot/pull/8986#discussion_r918204401
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -916,11 +930,15 @@ void updateIdealStateOnSegmentCompletion(String realtimeTableName, String commit
"Exceeded max segment completion time for segment " + committingSegmentName);
}
updateInstanceStatesForNewConsumingSegment(idealState.getRecord().getMapFields(), committingSegmentName,
- newSegmentName, segmentAssignment, instancePartitionsMap);
+ isTablePaused(idealState) ? null : newSegmentName, segmentAssignment, instancePartitionsMap);
Review Comment:
In regular cases, you're right. The new segment name in step 2 of the commit process will be null and here in step 3 we'll have a null newSegmentName. In an edge case where pause is issued after step 2 and right before step 3, the new segment name is not null, but we need to pass in null to prevent creating new consuming segment in ideal state.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] KKcorps commented on pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
KKcorps commented on PR #8986:
URL: https://github.com/apache/pinot/pull/8986#issuecomment-1190104911
@sajjad-moradi There seems to be test failure on this -
```
Error: PauseResumeConsumptionIntegrationTest.setUp:69->verifyPause:104->verify:116->assertCountStar:138 expected [9292] but found [5000]
[33317](https://github.com/apache/pinot/runs/7288357664?check_suite_focus=true#step:5:33318)
```
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] sajjad-moradi commented on pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on PR #8986:
URL: https://github.com/apache/pinot/pull/8986#issuecomment-1190758433
> @sajjad-moradi There seems to be test failure on this -
>
> ```
> Error: PauseResumeConsumptionIntegrationTest.setUp:69->verifyPause:104->verify:116->assertCountStar:138 expected [9292] but found [5000]
> [33317](https://github.com/apache/pinot/runs/7288357664?check_suite_focus=true#step:5:33318)
> ```
The test takes a long time. Disabling it for now. Added a comment there to revisit later to make the execution time shorter (maybe by producing less events).
The e2e functionality has been tested locally though, so it's not worth increasing the timeout which in turns lead to much higher build 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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] codecov-commenter commented on pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #8986:
URL: https://github.com/apache/pinot/pull/8986#issuecomment-1169406296
# [Codecov](https://codecov.io/gh/apache/pinot/pull/8986?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#8986](https://codecov.io/gh/apache/pinot/pull/8986?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (04e6270) into [master](https://codecov.io/gh/apache/pinot/commit/db07f2928a10907badd6b4c1ed583020e1b2d9d2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (db07f29) will **decrease** coverage by `2.28%`.
> The diff coverage is `9.02%`.
```diff
@@ Coverage Diff @@
## master #8986 +/- ##
============================================
- Coverage 28.83% 26.55% -2.29%
+ Complexity 47 1 -46
============================================
Files 1805 1807 +2
Lines 94438 94682 +244
Branches 14140 14160 +20
============================================
- Hits 27234 25142 -2092
- Misses 64654 67134 +2480
+ Partials 2550 2406 -144
```
| Flag | Coverage Δ | |
|---|---|---|
| integration1 | `26.55% <9.02%> (+0.03%)` | :arrow_up: |
| integration2 | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/pinot/pull/8986?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...ache/pinot/common/messages/ForceCommitMessage.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWVzc2FnZXMvRm9yY2VDb21taXRNZXNzYWdlLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...ot/common/protocols/SegmentCompletionProtocol.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vcHJvdG9jb2xzL1NlZ21lbnRDb21wbGV0aW9uUHJvdG9jb2wuamF2YQ==) | `92.38% <ø> (ø)` | |
| [...es/PinotControllerPeriodicTaskRestletResource.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9hcGkvcmVzb3VyY2VzL1Bpbm90Q29udHJvbGxlclBlcmlvZGljVGFza1Jlc3RsZXRSZXNvdXJjZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...ller/api/resources/PinotRealtimeTableResource.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9hcGkvcmVzb3VyY2VzL1Bpbm90UmVhbHRpbWVUYWJsZVJlc291cmNlLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...ntroller/helix/core/PinotHelixResourceManager.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL1Bpbm90SGVsaXhSZXNvdXJjZU1hbmFnZXIuamF2YQ==) | `36.44% <0.00%> (-2.48%)` | :arrow_down: |
| [.../helix/core/realtime/SegmentCompletionManager.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL3JlYWx0aW1lL1NlZ21lbnRDb21wbGV0aW9uTWFuYWdlci5qYXZh) | `45.52% <0.00%> (-1.32%)` | :arrow_down: |
| [...server/starter/helix/HelixInstanceDataManager.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zZXJ2ZXIvc3RhcnRlci9oZWxpeC9IZWxpeEluc3RhbmNlRGF0YU1hbmFnZXIuamF2YQ==) | `69.11% <0.00%> (-9.42%)` | :arrow_down: |
| [...er/starter/helix/SegmentMessageHandlerFactory.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zZXJ2ZXIvc3RhcnRlci9oZWxpeC9TZWdtZW50TWVzc2FnZUhhbmRsZXJGYWN0b3J5LmphdmE=) | `61.17% <0.00%> (-18.83%)` | :arrow_down: |
| [.../core/realtime/PinotLLCRealtimeSegmentManager.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL3JlYWx0aW1lL1Bpbm90TExDUmVhbHRpbWVTZWdtZW50TWFuYWdlci5qYXZh) | `55.85% <8.19%> (-6.72%)` | :arrow_down: |
| [...manager/realtime/LLRealtimeSegmentDataManager.java](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvTExSZWFsdGltZVNlZ21lbnREYXRhTWFuYWdlci5qYXZh) | `58.95% <12.50%> (-2.41%)` | :arrow_down: |
| ... and [221 more](https://codecov.io/gh/apache/pinot/pull/8986/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/pinot/pull/8986?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/pinot/pull/8986?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [db07f29...04e6270](https://codecov.io/gh/apache/pinot/pull/8986?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] mcvsubbu commented on pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on PR #8986:
URL: https://github.com/apache/pinot/pull/8986#issuecomment-1172793016
> > You should also check the idealstate once before entering the consumeLoop()
>
> That would be for handling a rare edge case. The probability of that case happening is very low and can be calculated as `t1/t2` where `t1`: time duration between receiving online-to-consuming msg and entering the consumeLoop `t2`: consumeLoop duration Note that t1 is in microseconds-milliseconds and t2 is in hours (depending on table config).
>
> If that case happens, the partition will eventually be paused when after the segment gets completed. The operator can reissue the pause request if they need it sooner.
Makes sense. Keeps the first version simpler while kicking the tires. We can enhance this as we hear back regarding its use.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] npawar commented on pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
npawar commented on PR #8986:
URL: https://github.com/apache/pinot/pull/8986#issuecomment-1229290663
@sajjad-moradi would you please add this to the documentation? (or point me to it if you already have, I couldn't find it easily). I think this would be a good place: https://docs.pinot.apache.org/basics/data-import/pinot-stream-ingestion, for all of them pause/resume/forceCommit/change stream.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] mcvsubbu commented on a diff in pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on code in PR #8986:
URL: https://github.com/apache/pinot/pull/8986#discussion_r912297405
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -49,27 +50,71 @@
HttpHeaders.AUTHORIZATION, in = ApiKeyAuthDefinition.ApiKeyLocation.HEADER, key = SWAGGER_AUTHORIZATION_KEY)))
@Path("/")
public class PinotRealtimeTableResource {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PinotRealtimeTableResource.class);
+
@Inject
PinotHelixResourceManager _pinotHelixResourceManager;
+ @Inject
+ PinotLLCRealtimeSegmentManager _pinotLLCRealtimeSegmentManager;
+
+ @POST
+ @Path("/tables/{tableName}/pauseConsumption")
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Pause consumption of a realtime table",
+ notes = "Pause the consumption of a realtime table")
+ public Response pauseConsumption(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName) {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ validate(tableNameWithType);
+ try {
+ return Response.ok(_pinotLLCRealtimeSegmentManager.pauseConsumption(tableNameWithType)).build();
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
+
@POST
@Path("/tables/{tableName}/resumeConsumption")
@Produces(MediaType.APPLICATION_JSON)
- @Consumes(MediaType.APPLICATION_JSON)
- @ApiOperation(value = "Resume the consumption of a realtime table",
- notes = "Resume the consumption of a realtime table")
- public String resumeConsumption(
- @ApiParam(value = "Name of the table", required = true)
- @PathParam("tableName") String tableName) throws JsonProcessingException {
- // TODO: Add util method for invoking periodic tasks
+ @ApiOperation(value = "Resume consumption of a realtime table",
+ notes = "Resume the consumption for a realtime table")
+ public Response resumeConsumption(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName) {
String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
- Map<String, String> taskProperties = new HashMap<>();
- taskProperties.put(RealtimeSegmentValidationManager.RECREATE_DELETED_CONSUMING_SEGMENT_KEY, "true");
+ validate(tableNameWithType);
+ try {
+ return Response.ok(_pinotLLCRealtimeSegmentManager.resumeConsumption(tableNameWithType)).build();
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
- Pair<String, Integer> taskExecutionDetails = _pinotHelixResourceManager
- .invokeControllerPeriodicTask(tableNameWithType, Constants.REALTIME_SEGMENT_VALIDATION_MANAGER, taskProperties);
+ @GET
+ @Path("/tables/{tableName}/pauseStatus")
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Return pause status of a realtime table",
+ notes = "Return pause status of a realtime table along with list of consuming segments.")
+ public Response getConsumptionStatus(
+ @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName) {
+ String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+ validate(tableNameWithType);
+ try {
+ return Response.ok().entity(_pinotLLCRealtimeSegmentManager.getPauseStatus(tableNameWithType)).build();
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+ }
- return "{\"Log Request Id\": \"" + taskExecutionDetails.getLeft()
- + "\",\"Controllers notified\":" + (taskExecutionDetails.getRight() > 0) + "}";
+ private void validate(String tableNameWithType) {
+ IdealState idealState = _pinotHelixResourceManager.getTableIdealState(tableNameWithType);
+ if (idealState == null) {
+ throw new ControllerApplicationException(LOGGER, "Ideal State is null for table " + tableNameWithType,
Review Comment:
Maybe say "table not found"
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -916,11 +930,15 @@ void updateIdealStateOnSegmentCompletion(String realtimeTableName, String commit
"Exceeded max segment completion time for segment " + committingSegmentName);
}
updateInstanceStatesForNewConsumingSegment(idealState.getRecord().getMapFields(), committingSegmentName,
- newSegmentName, segmentAssignment, instancePartitionsMap);
+ isTablePaused(idealState) ? null : newSegmentName, segmentAssignment, instancePartitionsMap);
Review Comment:
This method is called with `newSegmentName` set to null `null` if the table is being paused right? Why check idealstate again?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -3613,7 +3613,27 @@ public Pair<String, Integer> invokeControllerPeriodicTask(String tableName, Stri
LOGGER.info("[TaskRequestId: {}] Periodic task execution message sent to {} controllers.", periodicTaskRequestId,
messageCount);
- return Pair.of(periodicTaskRequestId, messageCount);
+ return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
+ }
+
+ public static class PeriodicTaskInvocationResponse {
Review Comment:
Same with this class. It is returned externally, move it so some api (or perhaps spi) package. Not sure what the convention is that we are following
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1399,4 +1417,120 @@ public void uploadToDeepStoreIfMissing(TableConfig tableConfig, List<SegmentZKMe
}
}
}
+
+ /**
+ * Pause consumption on a table by
+ * 1) setting "isTablePaused" in ideal states to true and
+ * 2) sending force commit messages to servers
+ */
+ public PauseStatus pauseConsumption(String tableNameWithType) {
+ IdealState updatedIdealState = updatePauseStatusInIdealState(tableNameWithType, true);
+ Set<String> consumingSegments = findConsumingSegments(updatedIdealState);
+ sendForceCommitMessageToServers(tableNameWithType, consumingSegments);
+ return new PauseStatus(true, consumingSegments, consumingSegments.isEmpty() ? null : "Pause flag is set."
Review Comment:
I would prefer returning a different class here than `PauseStatus`, since `PauseStatus` is an externally visible class, and we need to keep compatibility. For internal methods, we should be able to change things as needed
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1399,4 +1417,120 @@ public void uploadToDeepStoreIfMissing(TableConfig tableConfig, List<SegmentZKMe
}
}
}
+
+ /**
+ * Pause consumption on a table by
+ * 1) setting "isTablePaused" in ideal states to true and
+ * 2) sending force commit messages to servers
+ */
+ public PauseStatus pauseConsumption(String tableNameWithType) {
+ IdealState updatedIdealState = updatePauseStatusInIdealState(tableNameWithType, true);
+ Set<String> consumingSegments = findConsumingSegments(updatedIdealState);
+ sendForceCommitMessageToServers(tableNameWithType, consumingSegments);
+ return new PauseStatus(true, consumingSegments, consumingSegments.isEmpty() ? null : "Pause flag is set."
+ + " Consuming segments are being committed."
+ + " Use /pauseStatus endpoint in a few moments to check if all consuming segments have been committed.");
+ }
+
+ /**
+ * Resume consumption on a table by
+ * 1) setting "isTablePaused" in ideal states to false and
+ * 2) triggering segment validation job to create new consuming segments in ideal states
+ */
+ public PauseStatus resumeConsumption(String tableNameWithType) {
+ IdealState updatedIdealState = updatePauseStatusInIdealState(tableNameWithType, false);
+
+ // trigger realtime segment validation job to resume consumption
+ Map<String, String> taskProperties = new HashMap<>();
+ taskProperties.put(RealtimeSegmentValidationManager.RECREATE_DELETED_CONSUMING_SEGMENT_KEY, "true");
Review Comment:
create an issue to see if this is still needed. No that we have pause/resume, I see no need for a user to delete a consuming segment. We can disable such deletion in the api.
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java:
##########
@@ -169,6 +173,34 @@ public HelixTaskResult handleMessage()
}
}
+ private class ForceCommitMessageHandler extends DefaultMessageHandler {
+
+ private String _tableName;
+ private Set<String> _segmentNames;
+
+ public ForceCommitMessageHandler(ForceCommitMessage forceCommitMessage, ServerMetrics metrics,
+ NotificationContext ctx) {
+ super(forceCommitMessage, metrics, ctx);
+ _tableName = forceCommitMessage.getTableName();
+ _segmentNames = forceCommitMessage.getSegmentNames();
+ }
+
+ @Override
+ public HelixTaskResult handleMessage()
+ throws InterruptedException {
+ HelixTaskResult helixTaskResult = new HelixTaskResult();
+ _logger.info("Handling force commit message for table {} segments {}", _tableName, _segmentNames);
+ try {
+ _instanceDataManager.forceCommit(_tableName, _segmentNames);
+ helixTaskResult.setSuccess(true);
+ } catch (Exception e) {
+ _metrics.addMeteredTableValue(_tableNameWithType, ServerMeter.DELETE_TABLE_FAILURES, 1);
+ Utils.rethrowException(e);
Review Comment:
Not sure what helix does with exceptions thrown. Hopefully it does not retry.
Another way is to just log the exception and move on.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -552,12 +564,12 @@ private void commitSegmentMetadataInternal(String realtimeTableName,
.collect(Collectors.toSet());
int numPartitionGroups = newPartitionGroupMetadataList.size();
- // Only if committingSegment's partitionGroup is present in the newPartitionGroupMetadataList, we create new
- // segment metadata
String newConsumingSegmentName = null;
- String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
- long newSegmentCreationTimeMs = getCurrentTimeMs();
- if (newPartitionGroupSet.contains(committingSegmentPartitionGroupId)) {
+ if (!isTablePaused(idealState) && newPartitionGroupSet.contains(committingSegmentPartitionGroupId)) {
+ // Only if committingSegment's partitionGroup is present in the newPartitionGroupMetadataList, we create new
+ // segment metadata
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
Review Comment:
nice refactor moving these in
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/ControllerRequestClient.java:
##########
@@ -169,6 +170,39 @@ public void deleteSegments(String tableName, TableType tableType)
}
}
+ public PinotLLCRealtimeSegmentManager.PauseStatus pauseConsumption(String tableName)
+ throws IOException {
+ try {
+ SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPostRequest(new URL(
+ _controllerRequestURLBuilder.forPauseConsumption(tableName)).toURI(), null));
+ return JsonUtils.stringToObject(response.getResponse(), PinotLLCRealtimeSegmentManager.PauseStatus.class);
Review Comment:
We don't want `ControllerRequestClient` to depend on `PinotLLCRealtimeSegmentManager` right? Should we move `PauseStatus` to the controller/api package? We may rename or move the PinotLLCRealtmeSegmentManager at some point, and we don't want to rebuild clients just because. Perhaps there are other such uses in this file, maybe we should pay attention and fix those as well (on a separate PR/issue, of course) but we can stop introducing new ones
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] sajjad-moradi merged pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
sajjad-moradi merged PR #8986:
URL: https://github.com/apache/pinot/pull/8986
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] mcvsubbu commented on a diff in pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on code in PR #8986:
URL: https://github.com/apache/pinot/pull/8986#discussion_r918218555
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauseResumeConsumptionIntegrationTest.java:
##########
@@ -107,9 +107,8 @@ private void verifyResume(int numRecordsInTable) throws Exception {
}
private void verify(boolean pause, int expectedNumRecordsInTable) throws Exception {
- for (int i = 0; i < 30; i++) {
- PinotLLCRealtimeSegmentManager.PauseStatus pauseStatus =
- getControllerRequestClient().getPauseStatus(getTableName());
+ for (int i = 0; i < 60; i++) {
Review Comment:
can we declare 30/60/anytning as a final int since it is used in more than one place (and they need to be consistent)? thanks
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -109,11 +109,11 @@ public Response getConsumptionStatus(
private void validate(String tableNameWithType) {
IdealState idealState = _pinotHelixResourceManager.getTableIdealState(tableNameWithType);
if (idealState == null) {
- throw new ControllerApplicationException(LOGGER, "Ideal State is null for table " + tableNameWithType,
+ throw new ControllerApplicationException(LOGGER, String.format("Table %s not found!", tableNameWithType),
Response.Status.NOT_FOUND);
}
if (!idealState.isEnabled()) {
- throw new ControllerApplicationException(LOGGER, "Ideal State is disabled for table" + tableNameWithType,
+ throw new ControllerApplicationException(LOGGER, "Ideal State is disabled for table " + tableNameWithType,
Review Comment:
```suggestion
throw new ControllerApplicationException(LOGGER, "Disabled table: " + tableNameWithType,
```
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] sajjad-moradi commented on a diff in pull request #8986: Support pause/resume consumption of realtime tables
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on code in PR #8986:
URL: https://github.com/apache/pinot/pull/8986#discussion_r918208930
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/ControllerRequestClient.java:
##########
@@ -169,6 +170,39 @@ public void deleteSegments(String tableName, TableType tableType)
}
}
+ public PinotLLCRealtimeSegmentManager.PauseStatus pauseConsumption(String tableName)
+ throws IOException {
+ try {
+ SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPostRequest(new URL(
+ _controllerRequestURLBuilder.forPauseConsumption(tableName)).toURI(), null));
+ return JsonUtils.stringToObject(response.getResponse(), PinotLLCRealtimeSegmentManager.PauseStatus.class);
Review Comment:
Moved to api module.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -3613,7 +3613,27 @@ public Pair<String, Integer> invokeControllerPeriodicTask(String tableName, Stri
LOGGER.info("[TaskRequestId: {}] Periodic task execution message sent to {} controllers.", periodicTaskRequestId,
messageCount);
- return Pair.of(periodicTaskRequestId, messageCount);
+ return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
+ }
+
+ public static class PeriodicTaskInvocationResponse {
Review Comment:
Moved to api module.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org