healchow commented on code in PR #3567:
URL: https://github.com/apache/incubator-inlong/pull/3567#discussion_r845684992
##########
inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java:
##########
@@ -73,15 +73,15 @@ public ListenerResult listen(WorkflowContext context)
throws Exception {
new TypeReference<Map<String, String>>(){});
kvConf.putAll(result);
if (StringUtils.isEmpty(kvConf.get(InlongGroupSettings.SORT_JOB_ID))) {
- String message = String.format("inlongGroupId:%s not add
restartProcess listener,SORT_JOB_ID is empty",
+ String message = String.format("inlongGroupId:%s not add
restartProcess listener, SORT_JOB_ID is empty",
Review Comment:
Change to `"groupId [%s] not add restartProcess listener, as the sortJobId
is empty"`.
##########
inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/IntergrationTaskRunner.java:
##########
@@ -126,7 +101,7 @@ public void run() {
String jobId = flinkService.restore(flinkInfo);
log.info("Restore job {} success in backend",
jobId);
} catch (Exception e) {
- log.warn("Restore job failed in backend");
+ log.error("Restore job failed in backend",
e.getMessage());
Review Comment:
Should print stack traces of exception.
##########
inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java:
##########
@@ -64,7 +64,7 @@ public ListenerResult listen(WorkflowContext context) throws
Exception {
String dataFlows = kvConf.get(InlongGroupSettings.DATA_FLOW);
String inlongGroupId = context.getProcessForm().getInlongGroupId();
if (StringUtils.isEmpty(dataFlows)) {
- String message = String.format("inlongGroupId:%s not exec
startupProcess listener,dataflows is empty",
+ String message = String.format("inlongGroupId:%s not exec
startupProcess listener, dataflows is empty",
Review Comment:
Change to `groupId [%s] not exec startupProcess listener, as the dataflows
is empty`.
##########
inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java:
##########
@@ -261,169 +226,43 @@ 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) {
try {
- body = getReq(url);
- log.info("triggerSavepoint success");
- return body;
+ RestClusterClient<StandaloneClusterId> client = getFlinkClient();
+ JobID jobID = JobID.fromHexString(jobId);
+ CompletableFuture<String> stopResult =
+
client.stopWithSavepoint(jobID,requestBody.isDrain(),requestBody.getTargetDirectory());
+ String result = stopResult.get();
Review Comment:
Just use `return stopResult.get();`.
--
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]
