LvJiancheng commented on code in PR #3567:
URL: https://github.com/apache/incubator-inlong/pull/3567#discussion_r845188121
##########
inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java:
##########
@@ -108,7 +108,7 @@ public ListenerResult listen(WorkflowContext context)
throws Exception {
managerFlinkTask.start(flinkInfo);
log.info("the jobId {} submit success");
} catch (Exception e) {
- log.warn("startup exception [{}]", e.getMessage());
+ log.warn("startup exception [{}]", e.getMessage());
Review Comment:
done
##########
inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java:
##########
@@ -261,169 +228,45 @@ public String restore(FlinkInfo flinkInfo) {
return jobId;
}
} catch (Exception e) {
- log.error("submit job error", e);
+ log.error("restore job error: ", e);
}
return jobId;
}
/**
- * stop job
- * @param jobId
- */
- public ResponseBody stopJobs(String jobId, String requestBody) {
- ResponseBody responseBody = null;
- try {
- String httpUrl = urlHead + Constants.JOB_URL +
Constants.URL_SEPARATOR + jobId + Constants.SUSPEND_URL;
- OkHttpClient client = new OkHttpClient();
- MediaType mediaType = MediaType.parse("application/json;
charset=utf-8");
- RequestBody body = RequestBody.create(mediaType, requestBody);
- Request request = new Request.Builder()
- .url(httpUrl)
- .method("POST", body)
- .build();
- Response response = client.newCall(request).execute();
- log.info("the job : {} has stop",jobId);
- return response.body();
- } catch (Exception e) {
- log.error("stop job error", e.getMessage());
- }
- return responseBody;
- }
-
- /**
- * trigger savepoint
+ * stop job with savepoint
* @param jobId
- * @param triggerId
+ * @param requestBody
* @return
*/
- public ResponseBody triggerSavepoint(String jobId,String triggerId) {
- String url = urlHead + Constants.JOB_URL + Constants.URL_SEPARATOR +
jobId
- + Constants.SAVEPOINT + Constants.URL_SEPARATOR + triggerId;
- ResponseBody body = null;
+ public String stopJobs(String jobId, StopWithSavepointRequestBody
requestBody) {
+ String result = null;
try {
- body = getReq(url);
- log.info("triggerSavepoint success");
- return body;
- } catch (Exception e) {
- log.error("triggerSavepoint after stop job fail",e.getMessage());
- }
- return body;
- }
+ RestClusterClient<StandaloneClusterId> client = getFlinkClient();
+ JobID jobID = JobID.fromHexString(jobId);
+ CompletableFuture<String> stopResult =
+
client.stopWithSavepoint(jobID,requestBody.isDrain(),requestBody.getTargetDirectory());
+ result = stopResult.get();
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
