saurabhd336 commented on code in PR #10088:
URL: https://github.com/apache/pinot/pull/10088#discussion_r1066737744
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotRealtimeTableResource.java:
##########
@@ -116,29 +123,67 @@ public Response resumeConsumption(
@POST
@Path("/tables/{tableName}/forceCommit")
- @ApiOperation(value = "Force commit the current consuming segments",
- notes = "Force commit the current segments in consuming state and
restart consumption. "
+ @ApiOperation(value = "Force commit the current consuming segments", notes =
+ "Force commit the current segments in consuming state and restart
consumption. "
+ "This should be used after schema/table config changes. "
+ "Please note that this is an asynchronous operation, "
+ "and 200 response does not mean it has actually been done already")
- public Response forceCommit(
+ public SuccessResponse forceCommit(
@ApiParam(value = "Name of the table", required = true)
@PathParam("tableName") String tableName) {
String tableNameWithType =
TableNameBuilder.REALTIME.tableNameWithType(tableName);
validate(tableNameWithType);
+ String submittedJobId = null;
try {
- _pinotLLCRealtimeSegmentManager.forceCommit(tableNameWithType);
- return Response.ok().build();
+ Set<String> consumingSegmentsForceCommitted =
_pinotLLCRealtimeSegmentManager.forceCommit(tableNameWithType);
+ try {
+ String jobId = UUID.randomUUID().toString();
+ _pinotHelixResourceManager.addNewForceCommitJob(tableNameWithType,
jobId, consumingSegmentsForceCommitted);
+ submittedJobId = jobId;
+ } catch (Exception e) {
+ LOGGER.error("Could not add force commit job metadata to ZK table :
{}", tableNameWithType, e);
+ }
} catch (Exception e) {
throw new ControllerApplicationException(LOGGER, e.getMessage(),
Response.Status.INTERNAL_SERVER_ERROR, e);
}
+
+ return new SuccessResponse(
+ String.format("Successfully submitted force commit job id. Job meta ZK
storage status: %s, job id : %s",
+ (submittedJobId != null) ? "SUCCESS" : "FAILED", submittedJobId));
}
+ @GET
+ @Path("segments/forceCommitStatus/{jobId}")
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Get status for a submitted force commit operation",
notes = "Get status for a submitted "
+ + "force commit operation")
+ public Map<String, String> getForceCommitJobStatus(
+ @ApiParam(value = "Force commit job id", required = true)
@PathParam("jobId") String reloadJobId)
+ throws Exception {
+ Map<String, String> controllerJobZKMetadata =
_pinotHelixResourceManager.getControllerJobZKMetadata(reloadJobId);
+ String tableNameWithType =
controllerJobZKMetadata.get(CommonConstants.ControllerJob.TABLE_NAME_WITH_TYPE);
+ Set<String> consumingSegmentCommitted = new HashSet<>();
+ consumingSegmentCommitted = JsonUtils.stringToObject(
+
controllerJobZKMetadata.get(CommonConstants.ControllerJob.CONSUMING_SEGMENTS_FORCE_COMMITTED_LIST),
+ consumingSegmentCommitted.getClass());
+ Set<String> onlineSegmentsForTable =
+
_pinotHelixResourceManager.getOnlineSegmentsFromExternalView(tableNameWithType);
+ // Need to check how many are online
+ AtomicInteger onlineSegmentsCount = new AtomicInteger(0);
+ consumingSegmentCommitted.forEach(segmentName -> {
+ if (onlineSegmentsForTable.contains(segmentName)) {
+ onlineSegmentsCount.incrementAndGet();
+ }
+ });
+
+ controllerJobZKMetadata.put("onlineSegmentCount",
Integer.toString(onlineSegmentsCount.get()));
Review Comment:
Lambdas only accept effectively final variables.
--
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-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
