You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "J-HowHuang (via GitHub)" <gi...@apache.org> on 2026/01/13 00:28:04 UTC
[PR] Isolate potentially long-running minion task API resources on controller [pinot]
J-HowHuang opened a new pull request, #17494:
URL: https://github.com/apache/pinot/pull/17494
## Description
The current implementation of `PinotHelixTaskResourceManager` has a lot of synchronized methods, which introduces a potential global lock contention on the manager instance across all minion task related API resources (endpoints), see comments:
https://github.com/apache/pinot/blob/c23a1e971dceafae424ccd1d6c5b829ecf3c8e78/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java#L77-L81
For the API resources that call on any of these synchronized methods, it could become long-running if any other synchronized method is blocked. For example: https://github.com/apache/pinot/blob/c23a1e971dceafae424ccd1d6c5b829ecf3c8e78/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java#L284-L296
This method calls `_taskDriver.enqueueJob`, which writes to ZK with retries, and the retry mechanics is implemented by Helix. Some issues have been spotted with 24 Hrs with indefinite retry.
Now any call to `PinotHelixTaskResourceManager`'s synchronized methods will be blocked. Therefore requests to these API endpoint may block the thread pool that controller used to handle API request. In this case, this thread pool: https://github.com/apache/pinot/blob/8333688e4ecec030b334f4d3239b6737ef17fdea/pinot-core/src/main/java/org/apache/pinot/core/util/ListenerConfigUtil.java#L239-L241
This would result in a total unresponsive controller, even the request of pinot UI. Therefore we need to isolate these long-running API resources from other crucial API resources.
## Change
* Create a separate thread pool dedicated for potentially long-running minion task related API resources.
* The size of the thread pool is the same as our current http handler thread pool (`grizzly-http-server-%d`)
* Run the handler of these resources asynchronously, using Jersey's `@Suspended AsyncResponse` annotation.
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2691739687
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,49 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
+ _resourceExecutorService.execute(() -> {
+ try {
+ String translatedTableName = DatabaseUtils.translateTableName(tableName, headers);
+ TableType tableType = Constants.validateTableType(tableTypeStr);
+ if (tableType == null) {
+ asyncResponse.resume(
+ new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
+ Response.Status.BAD_REQUEST));
+ return;
+ }
+ String tableNameWithType =
+ ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, translatedTableName, tableType,
+ LOGGER).get(0);
+ String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
+ startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
+ startReplaceSegmentsRequest.getCustomMap());
+ asyncResponse.resume(
+ Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build());
+ } catch (Exception e) {
+ String tableNameWithType = null;
Review Comment:
Translation and validateTableType throws exception too. I've made the exception handling cleaner now.
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2700350657
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,45 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
+ _resourceExecutorService.execute(() -> {
+ String translatedTableName;
+ TableType tableType;
+ String tableNameWithType;
+ try {
+ translatedTableName = DatabaseUtils.translateTableName(tableName, headers);
Review Comment:
Tested, the exception is handled as sync response. Updated the exception handling on the three replacement API resources, now they're minimal change.
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on PR #17494:
URL: https://github.com/apache/pinot/pull/17494#issuecomment-3814482693
@Jackie-Jiang @krishan1390 Based on offline discussion, all the targeted API resources on the list are now annotated with `@ManagedAsync` to run in the cached thread pool.
And to monitor the active thread, these are the additional changes:
* Pinot controller now creates a cached thread pool with `@ManagedAsyncExecutor`. This thread pool is visible in Pinot and will replace Jersey's default cached thread pool. Now all API resources with `@ManagedAsync` will run in this cached thread pool instead.
* Add a new controller gauge `MANAGED_ASYNC_ACTIVE_THREADS` to report active threads in the cached thread pool.
The new gauge is tested locally:
<img width="641" height="491" alt="image" src="https://github.com/user-attachments/assets/a2be0a56-af01-442c-a464-7da8a983f800" />
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on PR #17494:
URL: https://github.com/apache/pinot/pull/17494#issuecomment-3762253710
> Do we want to run segment replacement related rest APIs using this new poll? They are not minion task resource, and they shouldn't hold lock similar to task related methods
@Jackie-Jiang These are heavily used by minion tasks. They're put within the same isolation because they're run by minion tasks. I don't think we need to further isolate them, doing so would be too fine-grained?
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2691618967
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,49 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
Review Comment:
This follows the examples from Jersey: https://eclipse-ee4j.github.io/jersey.github.io/documentation/latest/async.html
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2700348667
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,49 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
Review Comment:
removed `final` keyword
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2688669348
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,49 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
+ _resourceExecutorService.execute(() -> {
+ try {
+ String translatedTableName = DatabaseUtils.translateTableName(tableName, headers);
+ TableType tableType = Constants.validateTableType(tableTypeStr);
+ if (tableType == null) {
+ asyncResponse.resume(
+ new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
+ Response.Status.BAD_REQUEST));
+ return;
+ }
+ String tableNameWithType =
+ ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, translatedTableName, tableType,
+ LOGGER).get(0);
+ String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
+ startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
+ startReplaceSegmentsRequest.getCustomMap());
+ asyncResponse.resume(
+ Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build());
+ } catch (Exception e) {
+ String tableNameWithType = null;
Review Comment:
Does it have to be this complicated? Can we translate it in the main thread?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,49 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
Review Comment:
(nit) We don't usually put `final` in local variable
##########
pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java:
##########
@@ -618,6 +620,11 @@ private void setUpPinotController() {
_tenantRebalancer =
new TenantRebalancer(_tableRebalanceManager, _helixResourceManager, _rebalancerExecutorService);
+ // Set up executor service for specific resources for isolation
+ _minionTaskResourceExecutorService =
+ createExecutorService(HttpServerThreadPoolConfig.defaultInstance().getCorePoolSize(),
Review Comment:
Do you want to honor the `ListenerConfig` or find a different way to configure the pool size? E.g. we have a dedicate executor for rebalance
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang merged PR #17494:
URL: https://github.com/apache/pinot/pull/17494
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2755802710
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResource.java:
##########
@@ -190,9 +200,20 @@ public Set<String> listTaskTypes() {
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation("Get summary of all tasks across all task types, grouped by tenant. "
+ "Optionally filter by server tenant name to get tasks for a specific tenant only.")
- public PinotHelixTaskResourceManager.TaskSummaryResponse getTasksSummary(
+ @ManagedAsync
+ public void getTasksSummary(
@ApiParam(value = "Server tenant name to filter tasks. If not specified, returns all tenants grouped.")
- @QueryParam("tenant") @Nullable String tenantName) {
+ @QueryParam("tenant") @Nullable String tenantName,
+ @Suspended AsyncResponse asyncResponse) {
+ try {
+ PinotHelixTaskResourceManager.TaskSummaryResponse response = getTasksSummarySync(tenantName);
+ asyncResponse.resume(response);
+ } catch (Exception e) {
+ asyncResponse.resume(e);
+ }
+ }
+
+ private PinotHelixTaskResourceManager.TaskSummaryResponse getTasksSummarySync(@Nullable String tenantName) {
Review Comment:
(minor) Inline this. Same for other small sync calls
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResource.java:
##########
@@ -180,7 +180,17 @@ public class PinotTaskRestletResource {
@Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_TASK)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation("List all task types")
- public Set<String> listTaskTypes() {
+ @ManagedAsync
+ public void listTaskTypes(@Suspended AsyncResponse asyncResponse) {
+ try {
+ Set<String> taskTypes = listTaskTypesSync();
+ asyncResponse.resume(taskTypes);
+ } catch (Exception e) {
+ asyncResponse.resume(e);
+ }
+ }
+
+ private Set<String> listTaskTypesSync() {
Review Comment:
(minor) Inline this
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java:
##########
@@ -221,7 +221,8 @@ public enum ControllerGauge implements AbstractMetrics.Gauge {
// The progress of a certain table rebalance job of a table
TABLE_REBALANCE_JOB_PROGRESS_PERCENT("percent", false),
// HTTP thread utilization
- HTTP_THREAD_UTILIZATION("httpThreadUtilization", true);
+ HTTP_THREAD_UTILIZATION("httpThreadUtilization", true),
+ MANAGED_ASYNC_ACTIVE_THREADS("threads", true);
Review Comment:
（minor) Add comment for the intention of this gauge
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2691715742
##########
pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java:
##########
@@ -618,6 +620,11 @@ private void setUpPinotController() {
_tenantRebalancer =
new TenantRebalancer(_tableRebalanceManager, _helixResourceManager, _rebalancerExecutorService);
+ // Set up executor service for specific resources for isolation
+ _minionTaskResourceExecutorService =
+ createExecutorService(HttpServerThreadPoolConfig.defaultInstance().getCorePoolSize(),
Review Comment:
@krishan1390 suggests to make it the same size as current handler threads for now. Let me make a similar config to rebalance executor num_thread and default to this size.
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2691746556
##########
pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java:
##########
@@ -618,6 +620,11 @@ private void setUpPinotController() {
_tenantRebalancer =
new TenantRebalancer(_tableRebalanceManager, _helixResourceManager, _rebalancerExecutorService);
+ // Set up executor service for specific resources for isolation
+ _minionTaskResourceExecutorService =
+ createExecutorService(HttpServerThreadPoolConfig.defaultInstance().getCorePoolSize(),
Review Comment:
Add new conifg: `controller.executor.minionTaskResource.numThreads`, default value: `Runtime.getRuntime().availableProcessors() * 2`. This is the same value as `HttpServerThreadPoolConfig`'s default pool size
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2697237843
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,45 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
+ _resourceExecutorService.execute(() -> {
+ String translatedTableName;
+ TableType tableType;
+ String tableNameWithType;
+ try {
+ translatedTableName = DatabaseUtils.translateTableName(tableName, headers);
Review Comment:
When using `AsyncResponse`, does it work when the method throws exception? If it has the same behavior without `AsyncResponse`, we can keep the table translation and type check logic as is to simplify the code
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,49 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
Review Comment:
(minor) We follow our own convention. Also the ones added in `PinotTaskRestletResource` do not have `final` as well
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2687887446
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -984,29 +994,49 @@ public void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Start to replace segments", notes = "Start to replace segments")
- public Response startReplaceSegments(
+ public void startReplaceSegments(
@ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
@ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr,
@ApiParam(value = "Force cleanup") @QueryParam("forceCleanup") @DefaultValue("false") boolean forceCleanup,
@ApiParam(value = "Fields belonging to start replace segment request", required = true)
- StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers) {
- tableName = DatabaseUtils.translateTableName(tableName, headers);
- TableType tableType = Constants.validateTableType(tableTypeStr);
- if (tableType == null) {
- throw new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
- Response.Status.BAD_REQUEST);
- }
- String tableNameWithType =
- ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
- try {
- String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
- startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
- startReplaceSegmentsRequest.getCustomMap());
- return Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build();
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
- throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
- }
+ StartReplaceSegmentsRequest startReplaceSegmentsRequest, @Context HttpHeaders headers,
+ @Suspended final AsyncResponse asyncResponse) {
+ _resourceExecutorService.execute(() -> {
+ try {
+ String translatedTableName = DatabaseUtils.translateTableName(tableName, headers);
+ TableType tableType = Constants.validateTableType(tableTypeStr);
+ if (tableType == null) {
+ asyncResponse.resume(
+ new ControllerApplicationException(LOGGER, "Table type should either be offline or realtime",
+ Response.Status.BAD_REQUEST));
+ return;
+ }
+ String tableNameWithType =
+ ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, translatedTableName, tableType,
+ LOGGER).get(0);
+ String segmentLineageEntryId = _pinotHelixResourceManager.startReplaceSegments(tableNameWithType,
+ startReplaceSegmentsRequest.getSegmentsFrom(), startReplaceSegmentsRequest.getSegmentsTo(), forceCleanup,
+ startReplaceSegmentsRequest.getCustomMap());
+ asyncResponse.resume(
+ Response.ok(JsonUtils.newObjectNode().put("segmentLineageEntryId", segmentLineageEntryId)).build());
+ } catch (Exception e) {
+ String tableNameWithType = null;
+ try {
+ String translatedTableName = DatabaseUtils.translateTableName(tableName, headers);
+ TableType tableType = Constants.validateTableType(tableTypeStr);
+ if (tableType != null) {
+ tableNameWithType =
+ ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, translatedTableName, tableType,
+ LOGGER).get(0);
+ _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.NUMBER_START_REPLACE_FAILURE, 1);
+ }
+ } catch (Exception ignored) {
+ // Ignore errors when trying to get table name for metrics
+ }
Review Comment:
Duplicated error handling logic appears in three methods (startReplaceSegments, endReplaceSegments, revertReplaceSegments) at lines 1023-1035, 1080-1092, and 1134-1147. This repetitive code should be extracted into a helper method that takes the operation name (for the appropriate ControllerMeter constant) as a parameter.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResourceTest.java:
##########
@@ -70,6 +81,12 @@ public void init()
throws URISyntaxException {
MockitoAnnotations.openMocks(this);
when(_uriInfo.getRequestUri()).thenReturn(new URI("http://localhost:9000"));
+ // Make executor service execute tasks synchronously for testing
+ doAnswer(invocation -> {
+ Runnable runnable = invocation.getArgument(0);
+ runnable.run();
+ return null;
+ }).when(_minionTaskResourceExecutorService).execute(any(Runnable.class));
Review Comment:
The test setup forces synchronous execution of async tasks, which doesn't verify the actual async behavior. While this simplifies testing, it fails to validate that the executor service is properly handling task submission, exception propagation, and concurrent execution scenarios that could occur in production.
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "krishan1390 (via GitHub)" <gi...@apache.org>.
krishan1390 commented on PR #17494:
URL: https://github.com/apache/pinot/pull/17494#issuecomment-3742002406
@Jackie-Jiang this will change the default behaviour in OSS. I am assuming that is fine as there isn't another way to do this I think.
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #17494:
URL: https://github.com/apache/pinot/pull/17494#issuecomment-3741328541
### :warning: Unsupported file format
<details><summary>Upload processing failed due to unsupported file format. Please review the parser error message:</summary>
<pre><code class="language-python">Error parsing JUnit XML in /home/runner/work/pinot/pinot/pinot-query-runtime/target/surefire-reports/TEST-org.apache.pinot.query.runtime.queries.ResourceBasedQueriesTest.xml at 178:1069
Caused by:
RuntimeError: Error converting computed name to ValidatedString
Caused by:
string is too long</code></pre>
</details>
For more help, visit our [troubleshooting guide](https://docs.codecov.com/docs/test-analytics#troubleshooting).
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2756552156
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResource.java:
##########
@@ -190,9 +200,20 @@ public Set<String> listTaskTypes() {
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation("Get summary of all tasks across all task types, grouped by tenant. "
+ "Optionally filter by server tenant name to get tasks for a specific tenant only.")
- public PinotHelixTaskResourceManager.TaskSummaryResponse getTasksSummary(
+ @ManagedAsync
+ public void getTasksSummary(
@ApiParam(value = "Server tenant name to filter tasks. If not specified, returns all tenants grouped.")
- @QueryParam("tenant") @Nullable String tenantName) {
+ @QueryParam("tenant") @Nullable String tenantName,
+ @Suspended AsyncResponse asyncResponse) {
+ try {
+ PinotHelixTaskResourceManager.TaskSummaryResponse response = getTasksSummarySync(tenantName);
+ asyncResponse.resume(response);
+ } catch (Exception e) {
+ asyncResponse.resume(e);
+ }
+ }
+
+ private PinotHelixTaskResourceManager.TaskSummaryResponse getTasksSummarySync(@Nullable String tenantName) {
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #17494:
URL: https://github.com/apache/pinot/pull/17494#issuecomment-3838018680
Seems tests are running for much longer, and being cancelled after 90 minutes
--
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
Re: [PR] Isolate potentially long-running minion task API resources on controller [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17494:
URL: https://github.com/apache/pinot/pull/17494#discussion_r2755794798
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResource.java:
##########
@@ -672,71 +888,83 @@ public Map<String, Object> getCronSchedulerJobDetails(
@ApiOperation("Schedule tasks and return a map from task type to task name scheduled. If task type is missing, "
+ "schedules all tasks. If table name is missing, schedules tasks for all tables in the database. If database "
+ "is missing in headers, uses default.")
- @Nullable
- public Map<String, String> scheduleTasks(
+ @ManagedAsync
+ public void scheduleTasks(
@ApiParam(value = "Task type. If missing, schedules all tasks.") @QueryParam("taskType") @Nullable
String taskType,
@ApiParam(value = "Table name (with type suffix). If missing, schedules tasks for all tables in the database.")
@QueryParam("tableName") @Nullable String tableName,
@ApiParam(value = "Minion Instance tag to schedule the task explicitly on") @QueryParam("minionInstanceTag")
- @Nullable String minionInstanceTag, @Context HttpHeaders headers) {
- String database = headers != null ? headers.getHeaderString(DATABASE) : DEFAULT_DATABASE;
- Map<String, String> response = new HashMap<>();
- List<String> generationErrors = new ArrayList<>();
- List<String> schedulingErrors = new ArrayList<>();
- TaskSchedulingContext context = new TaskSchedulingContext()
- .setTriggeredBy(CommonConstants.TaskTriggers.MANUAL_TRIGGER.name())
- .setMinionInstanceTag(minionInstanceTag)
- .setLeader(false);
- if (taskType != null) {
- context.setTasksToSchedule(Collections.singleton(taskType));
- }
- if (tableName != null) {
- context.setTablesToSchedule(Collections.singleton(DatabaseUtils.translateTableName(tableName, headers)));
- } else {
- context.setDatabasesToSchedule(Collections.singleton(database));
- }
- Map<String, TaskSchedulingInfo> allTaskInfos = _pinotTaskManager.scheduleTasks(context);
- allTaskInfos.forEach((key, value) -> {
- if (value.getScheduledTaskNames() != null) {
- response.put(key, String.join(",", value.getScheduledTaskNames()));
+ @Nullable String minionInstanceTag, @Context HttpHeaders headers, @Suspended AsyncResponse asyncResponse) {
+ try {
+ String database = headers != null ? headers.getHeaderString(DATABASE) : DEFAULT_DATABASE;
+ Map<String, String> response = new HashMap<>();
+ List<String> generationErrors = new ArrayList<>();
+ List<String> schedulingErrors = new ArrayList<>();
+ TaskSchedulingContext context = new TaskSchedulingContext()
+ .setTriggeredBy(CommonConstants.TaskTriggers.MANUAL_TRIGGER.name())
+ .setMinionInstanceTag(minionInstanceTag)
+ .setLeader(false);
+ if (taskType != null) {
+ context.setTasksToSchedule(Collections.singleton(taskType));
}
- generationErrors.addAll(value.getGenerationErrors());
- schedulingErrors.addAll(value.getSchedulingErrors());
- });
- response.put(GENERATION_ERRORS_KEY, String.join(",", generationErrors));
- response.put(SCHEDULING_ERRORS_KEY, String.join(",", schedulingErrors));
- return response;
+ if (tableName != null) {
+ context.setTablesToSchedule(Collections.singleton(DatabaseUtils.translateTableName(tableName, headers)));
+ } else {
+ context.setDatabasesToSchedule(Collections.singleton(database));
+ }
+ Map<String, TaskSchedulingInfo> allTaskInfos = _pinotTaskManager.scheduleTasks(context);
+ allTaskInfos.forEach((key, value) -> {
+ if (value.getScheduledTaskNames() != null) {
+ response.put(key, String.join(",", value.getScheduledTaskNames()));
+ }
+ generationErrors.addAll(value.getGenerationErrors());
+ schedulingErrors.addAll(value.getSchedulingErrors());
+ });
+ response.put(GENERATION_ERRORS_KEY, String.join(",", generationErrors));
+ response.put(SCHEDULING_ERRORS_KEY, String.join(",", schedulingErrors));
+ asyncResponse.resume(response);
+ } catch (Exception e) {
+ asyncResponse.resume(new ControllerApplicationException(LOGGER,
+ String.format("Failed to schedule tasks due to error: %s", ExceptionUtils.getStackTrace(e)),
+ Response.Status.INTERNAL_SERVER_ERROR, e));
+ }
}
@POST
- @ManagedAsync
@Produces(MediaType.APPLICATION_JSON)
@Path("/tasks/execute")
@Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.EXECUTE_TASK)
@Authenticate(AccessType.CREATE)
@ApiOperation("Execute a task on minion")
+ @ManagedAsync
public void executeAdhocTask(AdhocTaskConfig adhocTaskConfig, @Suspended AsyncResponse asyncResponse,
@Context Request requestContext) {
try {
- asyncResponse.resume(_pinotTaskManager.createTask(adhocTaskConfig.getTaskType(), adhocTaskConfig.getTableName(),
- adhocTaskConfig.getTaskName(), adhocTaskConfig.getTaskConfigs()));
- } catch (TableNotFoundException e) {
- throw new ControllerApplicationException(LOGGER, "Failed to find table: " + adhocTaskConfig.getTableName(),
- Response.Status.NOT_FOUND, e);
- } catch (TaskAlreadyExistsException e) {
- throw new ControllerApplicationException(LOGGER, "Task already exists: " + adhocTaskConfig.getTaskName(),
- Response.Status.CONFLICT, e);
- } catch (UnknownTaskTypeException e) {
- throw new ControllerApplicationException(LOGGER, "Unknown task type: " + adhocTaskConfig.getTaskType(),
- Response.Status.NOT_FOUND, e);
- } catch (NoTaskScheduledException e) {
- throw new ControllerApplicationException(LOGGER,
- "No task is generated for table: " + adhocTaskConfig.getTableName() + ", with task type: "
- + adhocTaskConfig.getTaskType(), Response.Status.BAD_REQUEST, e);
+ try {
+ asyncResponse.resume(
+ _pinotTaskManager.createTask(adhocTaskConfig.getTaskType(), adhocTaskConfig.getTableName(),
Review Comment:
The `executeAdhocTask` method has nested try-catch blocks that reduce readability. The outer try-catch at line 943 catches all exceptions and calls `asyncResponse.resume(e)`, while the inner try-catch at line 944 catches specific exceptions and throws `ControllerApplicationException`. Since the outer catch will handle the exceptions thrown by the inner block, consider consolidating these into a single try-catch structure or removing the outer try-catch if the inner exception handling is sufficient.
--
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