Force commit consuming segments (#9197)

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

@@ -90,6 +90,26 @@ 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. "
+          + "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(
+      @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName) {
+    String tableNameWithType = TableNameBuilder.REALTIME.tableNameWithType(tableName);
+    validate(tableNameWithType);
+    try {
+      _pinotLLCRealtimeSegmentManager.forceCommit(tableNameWithType);
+      return Response.ok().build();
+    } catch (Exception e) {
+      throw new ControllerApplicationException(LOGGER, e.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, e);
+    }
+  }
+
+
   @GET
   @Path("/tables/{tableName}/pauseStatus")
   @Produces(MediaType.APPLICATION_JSON)

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

@@ -1420,6 +1420,15 @@ public void uploadToDeepStoreIfMissing(TableConfig tableConfig, List<SegmentZKMe
     }
   }
 
+  /**
+   * Force commit the current segments in consuming state and restart consumption
+   */
+  public void forceCommit(String tableNameWithType) {
+    IdealState idealState = getIdealState(tableNameWithType);
+    Set<String> consumingSegments = findConsumingSegments(idealState);
+    sendForceCommitMessageToServers(tableNameWithType, consumingSegments);
+  }
+
   /**
    * Pause consumption on a table by
    *   1) setting "isTablePaused" in ideal states to true and