healchow commented on code in PR #3567:
URL: https://github.com/apache/incubator-inlong/pull/3567#discussion_r845046494
##########
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();
+ return result;
- /**
- *
- * @return
- */
- public ResponseBody listJars() {
- String url = urlHead + Constants.JARS_URL;
- ResponseBody body = null;
- try {
- body = getReq(url);
- log.info("all jarfileinfo : {}",body);
- return body;
} catch (Exception e) {
- log.error("fetch jars fail",e.getMessage());
- }
- return body;
- }
-
- /**
- * upload jar
- * @param jarPath
- * @return
- * @throws Exception
- */
- public ResponseBody uploadJar(String jarPath,String fileName) throws
Exception {
- String url = urlHead + Constants.JARS_URL + Constants.UPLOAD;
- OkHttpClient client = new OkHttpClient();
- RequestBody requestBody = new MultipartBody.Builder()
- .setType(MultipartBody.FORM)
- .addFormDataPart("file", fileName,
-
RequestBody.create(MediaType.parse("multipart/form-data"), new File(jarPath)))
- .build();
-
- Request request = new Request.Builder()
- .header("Authorization", "Client-ID " + UUID.randomUUID())
- .url(url)
- .post(requestBody)
- .build();
-
- Response response = client.newCall(request).execute();
- if (!response.isSuccessful()) {
- throw new IOException("Unexpected code " + response);
- }
- return response.body();
- }
-
- /**
- * submit Jobs Without Jar
- * @param jarId
- * @param jarRunRequestbody
- * @return
- * @throws JsonProcessingException
- */
- public ResponseBody submitJobsWithoutJar(String jarId,JarRunRequestbody
jarRunRequestbody)
- throws JsonProcessingException {
- String httpUrl = urlHead + Constants.JARS_URL + jarId +
Constants.RUN_URL;
- ObjectMapper objectMapper = new ObjectMapper();
- String requestBody =
objectMapper.writeValueAsString(jarRunRequestbody);
- Response response = null;
- 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();
- try {
- response = client.newCall(request).execute();
- log.info("send remote req {}", response.body().string());
- return response.body();
- } catch (IOException e) {
- log.warn("send remote req fail {}", e.getMessage());
+ log.error("stop job error: ", e);
}
- return response.body();
+ return result;
}
/**
- * delete job
+ * cancel job
* @param jobId
+ * @return
*/
- public void deleteJobs(String jobId) {
- String url = urlHead + Constants.JOB_URL + Constants.URL_SEPARATOR +
jobId;
- OkHttpClient client = new OkHttpClient();
- MediaType mediaType = MediaType.parse("application/json;
charset=utf-8");
- RequestBody body = RequestBody.create(mediaType, "");
- Request request = new Request.Builder()
- .url(url)
- .method("PATCH", body)
- .build();
- try {
- client.newCall(request).execute();
- log.info("the jobId:{} delete successfuly",jobId);
-
- } catch (IOException e) {
- log.warn("the jobId:{} delete failure",jobId);
- }
- }
-
- /**
- * delete jar
- * @param jarId
- */
- public void deleteJars(String jarId) {
- String url = urlHead + Constants.JARS_URL + jarId;
- OkHttpClient client = new OkHttpClient();
- RequestBody body = null;
- Request request = new Request.Builder()
- .url(url)
- .method("DELETE", body)
- .build();
+ public void cancelJobs(String jobId) {
try {
- client.newCall(request).execute();
- log.info("the jobId:{} delete successfuly",jarId);
- } catch (IOException e) {
- log.warn("the jobId:{} delete successfuly",jarId);
+ RestClusterClient<StandaloneClusterId> client = getFlinkClient();
+ JobID jobID = JobID.fromHexString(jobId);
+ CompletableFuture<Acknowledge> result = client.cancel(jobID);
Review Comment:
What is the purpose of the `result`?
##########
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:
Change to `log.warn("startup exception: ", e);`.
##########
inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/SuspendSortListener.java:
##########
@@ -59,7 +59,7 @@ public ListenerResult listen(WorkflowContext context) throws
Exception {
InlongGroupExtInfo::getKeyName,
InlongGroupExtInfo::getKeyValue));
String sortExt = kvConf.get(InlongGroupSettings.SORT_PROPERTIES);
if (StringUtils.isEmpty(sortExt)) {
- String message = String.format("inlongGroupId:%s not add
suspendProcess listener,sortProperties is empty",
+ String message = String.format("inlongGroupId:%s not add
suspendProcess listener, sortProperties is empty",
Review Comment:
Change the message to `String.format("groupId: %s not add suspendProcess
listener, as the sortProperties is empty"`.
##########
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:
The local variable 'result' is redundant, just inline variable.
--
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]
