This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 6b64def23 [INLONG-3565][Manager] Unified the interface for Flink
plugin (#3567)
6b64def23 is described below
commit 6b64def23d1b23d2a811df15857e28178c52d37e
Author: jiancheng Lv <[email protected]>
AuthorDate: Fri Apr 8 11:20:48 2022 +0800
[INLONG-3565][Manager] Unified the interface for Flink plugin (#3567)
---
.../inlong/manager/plugin/flink/FlinkService.java | 233 +++------------------
.../plugin/flink/IntergrationTaskRunner.java | 67 +-----
.../manager/plugin/flink/ManagerFlinkTask.java | 4 +-
.../plugin/listener/DeleteSortListener.java | 7 +-
.../plugin/listener/RestartSortListener.java | 16 +-
.../plugin/listener/StartupSortListener.java | 12 +-
.../plugin/listener/SuspendSortListener.java | 6 +-
.../plugin/listener/SuspendSortListenerTest.java | 2 +-
8 files changed, 61 insertions(+), 286 deletions(-)
diff --git
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java
index 23189e977..6dbe835e0 100644
---
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java
+++
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java
@@ -17,16 +17,7 @@
package org.apache.inlong.manager.plugin.flink;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
-import okhttp3.MediaType;
-import okhttp3.MultipartBody;
-import okhttp3.OkHttpClient;
-import okhttp3.Request;
-import okhttp3.RequestBody;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
@@ -43,7 +34,7 @@ import
org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.plugin.flink.dto.FlinkConfig;
import org.apache.inlong.manager.plugin.flink.dto.FlinkInfo;
-import org.apache.inlong.manager.plugin.flink.dto.JarRunRequestbody;
+import org.apache.inlong.manager.plugin.flink.dto.StopWithSavepointRequestBody;
import org.apache.inlong.manager.plugin.util.FlinkConfiguration;
import java.io.File;
@@ -52,7 +43,6 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -66,7 +56,6 @@ public class FlinkService {
private final Integer port;
private final Integer jobManagerPort;
private final String address;
- private final String urlHead;
private final Integer parallelism;
private final String savepointDirectory;
@@ -83,7 +72,6 @@ public class FlinkService {
address = translateFromEndpont(endpoint).get("address");
port = Integer.valueOf(translateFromEndpont(endpoint).get("port"));
}
- urlHead = Constants.HTTP_URL + address + Constants.SEPARATOR + port;
}
/**
@@ -102,7 +90,7 @@ public class FlinkService {
return map;
}
} catch (Exception e) {
- log.error("fetch addres:port fail", e.getMessage());
+ log.error("fetch address:port fail: ", e);
}
return map;
}
@@ -141,45 +129,20 @@ public class FlinkService {
}
- /**
- * get request
- * @param httpurl
- * @return
- */
- public ResponseBody getReq(String httpurl) {
- Response response = null;
- OkHttpClient client = new OkHttpClient();
- Request request = new Request.Builder()
- .header("Authorization", "Client-ID " + UUID.randomUUID())
- .url(httpurl)
- .build();
- try {
- response = client.newCall(request).execute();
- log.info("get req {}",response.body());
- return response.body();
- } catch (IOException e) {
- log.warn("get req fail {}", e.getMessage());
- }
- return response.body();
- }
-
/**
* get job status
* @return
*/
public JobStatus getJobStatus(String jobId) {
- JobStatus status = null;
try {
RestClusterClient<StandaloneClusterId> client = getFlinkClient();
JobID jobID = JobID.fromHexString(jobId);
CompletableFuture<JobStatus> jobStatus =
client.getJobStatus(jobID);
- status = jobStatus.get();
- return status;
-
+ return jobStatus.get();
} catch (Exception e) {
- e.printStackTrace();
+ log.error("get job status error: ", e);
}
- return status;
+ return null;
}
/**
@@ -187,18 +150,15 @@ public class FlinkService {
* @return
*/
public JobDetailsInfo getJobDetail(String jobId) {
- JobDetailsInfo jobDetailsInfo = null;
try {
RestClusterClient<StandaloneClusterId> client = getFlinkClient();
JobID jobID = JobID.fromHexString(jobId);
CompletableFuture<JobDetailsInfo> jobDetails =
client.getJobDetails(jobID);
- jobDetailsInfo = jobDetails.get();
- return jobDetailsInfo;
-
+ return jobDetails.get();
} catch (Exception e) {
- e.printStackTrace();
+ log.error("get job detail error: ", e);
}
- return jobDetailsInfo;
+ return null;
}
/**
@@ -209,7 +169,6 @@ public class FlinkService {
RestClusterClient<StandaloneClusterId> client = null;
String localJarPath = flinkInfo.getLocalJarPath();
String[] programArgs = genProgramArgs(flinkInfo);
- String jobId = "";
try {
client = getFlinkClient();
Configuration configuration = initConfiguration();
@@ -224,12 +183,11 @@ public class FlinkService {
JobGraph jobGraph =
PackagedProgramUtils.createJobGraph(program,configuration,parallelism,false);
CompletableFuture<JobID> result = client.submitJob(jobGraph);
- jobId = result.get().toString();
- return jobId;
+ return result.get().toString();
} catch (Exception e) {
- log.error("submit job error", e);
+ log.error("submit job error: ", e);
}
- return jobId;
+ return null;
}
/**
@@ -240,7 +198,6 @@ public class FlinkService {
RestClusterClient<StandaloneClusterId> client = null;
String localJarPath = flinkInfo.getLocalJarPath();
String[] programArgs = genProgramArgs(flinkInfo);
- String jobId = "";
try {
client = getFlinkClient();
Configuration configuration = initConfiguration();
@@ -257,173 +214,45 @@ public class FlinkService {
JobGraph jobGraph =
PackagedProgramUtils.createJobGraph(program,configuration,parallelism,false);
CompletableFuture<JobID> result = client.submitJob(jobGraph);
- jobId = result.get().toString();
- return jobId;
+ return result.get().toString();
}
} catch (Exception e) {
- log.error("submit job error", e);
+ log.error("restore job error: ", e);
}
- return jobId;
+ return null;
}
/**
- * stop job
+ * stop job with savepoint
* @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
- * @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());
+ return stopResult.get();
} catch (Exception e) {
- log.error("triggerSavepoint after stop job fail",e.getMessage());
+ log.error("stop job error: ", e);
}
- return body;
+ return null;
}
/**
- *
+ * cancel job
+ * @param jobId
* @return
*/
- public ResponseBody listJars() {
- String url = urlHead + Constants.JARS_URL;
- ResponseBody body = null;
+ public void cancelJobs(String jobId) {
try {
- body = getReq(url);
- log.info("all jarfileinfo : {}",body);
- return body;
+ RestClusterClient<StandaloneClusterId> client = getFlinkClient();
+ JobID jobID = JobID.fromHexString(jobId);
+ client.cancel(jobID);
} 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());
- }
- return response.body();
- }
-
- /**
- * delete job
- * @param jobId
- */
- 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();
- try {
- client.newCall(request).execute();
- log.info("the jobId:{} delete successfuly",jarId);
- } catch (IOException e) {
- log.warn("the jobId:{} delete successfuly",jarId);
+ log.error("cancel job error: ", e);
}
}
diff --git
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/IntergrationTaskRunner.java
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/IntergrationTaskRunner.java
index 3516304db..7399933de 100644
---
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/IntergrationTaskRunner.java
+++
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/IntergrationTaskRunner.java
@@ -17,21 +17,13 @@
package org.apache.inlong.manager.plugin.flink;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
-import okhttp3.ResponseBody;
-import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.JobStatus;
import org.apache.inlong.manager.plugin.flink.dto.FlinkConfig;
import org.apache.inlong.manager.plugin.flink.dto.FlinkInfo;
import org.apache.inlong.manager.plugin.flink.dto.StopWithSavepointRequestBody;
import org.apache.inlong.manager.plugin.flink.enums.TaskCommitType;
-import java.util.HashMap;
-import java.util.Map;
-
import static org.apache.flink.api.common.JobStatus.FINISHED;
import static
org.apache.inlong.manager.plugin.util.FlinkUtils.getExceptionStackMsg;
@@ -94,29 +86,12 @@ public class IntergrationTaskRunner implements Runnable {
break;
case RESTART:
try {
- ObjectMapper objectMapper = new ObjectMapper();
StopWithSavepointRequestBody stopWithSavepointRequestBody
= new StopWithSavepointRequestBody();
stopWithSavepointRequestBody.setDrain(Constants.DRAIN);
stopWithSavepointRequestBody.setTargetDirectory(Constants.SAVEPOINT_DIRECTORY);
- String requestBody =
objectMapper.writeValueAsString(stopWithSavepointRequestBody);
- ResponseBody body =
flinkService.stopJobs(flinkInfo.getJobId(), requestBody);
- Map<String, String> map =
- objectMapper.readValue(body.string(), new
TypeReference<HashMap<String, String>>() {
- });
- String triggerId = map.get("request-id");
- if (StringUtils.isNotEmpty(triggerId)) {
- ResponseBody responseBody =
flinkService.triggerSavepoint(flinkInfo.getJobId(), triggerId);
- Map<String, JsonNode> dataflowMap =
-
objectMapper.convertValue(objectMapper.readTree(responseBody.string()),
- new TypeReference<Map<String,
JsonNode>>() {
- });
- JsonNode jobStopStatus = dataflowMap.get("status");
- JsonNode operation = dataflowMap.get("operation");
- String status = jobStopStatus.get("id").asText();
- String savepointPath =
operation.get("location").asText();
- flinkInfo.setSavepointPath(savepointPath);
- log.info("the jobId :{} status: {} ",
flinkInfo.getJobId(), status);
- }
+ String location =
flinkService.stopJobs(flinkInfo.getJobId(), stopWithSavepointRequestBody);
+ flinkInfo.setSavepointPath(location);
+ log.info("the jobId: {} savepoint: {} ",
flinkInfo.getJobId(), location);
int times = 0;
while (times < TRY_MAX_TIMES) {
JobStatus jobStatus =
flinkService.getJobStatus(flinkInfo.getJobId());
@@ -126,7 +101,7 @@ public class IntergrationTaskRunner implements Runnable {
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);
}
break;
}
@@ -149,35 +124,13 @@ public class IntergrationTaskRunner implements Runnable {
break;
case STOP:
try {
- ObjectMapper objectMapper = new ObjectMapper();
StopWithSavepointRequestBody stopWithSavepointRequestBody
= new StopWithSavepointRequestBody();
stopWithSavepointRequestBody.setDrain(Constants.DRAIN);
FlinkConfig flinkConfig = flinkService.getFlinkConfig();
stopWithSavepointRequestBody.setTargetDirectory(flinkConfig.getSavepointDirectory());
- String requestBody =
objectMapper.writeValueAsString(stopWithSavepointRequestBody);
- ResponseBody body =
flinkService.stopJobs(flinkInfo.getJobId(), requestBody);
- Map<String, String> map =
- objectMapper.readValue(body.string(), new
TypeReference<HashMap<String, String>>() {
- });
- String triggerId = map.get("request-id");
- if (StringUtils.isNotEmpty(triggerId)) {
- ResponseBody responseBody =
flinkService.triggerSavepoint(flinkInfo.getJobId(),triggerId);
- Map<String, JsonNode> dataflowMap =
-
objectMapper.convertValue(objectMapper.readTree(responseBody.string()),
- new TypeReference<Map<String, JsonNode>>(){});
- JsonNode jobStatus = dataflowMap.get("status");
- JsonNode operation = dataflowMap.get("operation");
- String status = jobStatus.get("id").asText();
- String savepointPath =
operation.get("location").asText();
- flinkInfo.setSavepointPath(savepointPath);
- log.info("the jobId {} status: {} ",
flinkInfo.getJobId(),status);
- }
- JobStatus jobStatus =
flinkService.getJobStatus(flinkInfo.getJobId());
- if (jobStatus.isTerminalState()) {
- log.info("stop job {}, status: {}, success in
backend", flinkInfo.getJobId(), jobStatus);
- } else {
- log.info("stop job {}, status: {}, fail in backend",
flinkInfo.getJobId(), jobStatus);
- }
+ String location =
flinkService.stopJobs(flinkInfo.getJobId(), stopWithSavepointRequestBody);
+ flinkInfo.setSavepointPath(location);
+ log.info("the jobId {} savepoint: {} ",
flinkInfo.getJobId(), location);
} catch (Exception e) {
String msg = String.format("stop job %s failed in backend
exception[%s]", flinkInfo.getJobId(),
getExceptionStackMsg(e));
@@ -188,13 +141,13 @@ public class IntergrationTaskRunner implements Runnable {
break;
case DELETE:
try {
- flinkService.deleteJobs(flinkInfo.getJobId());
+ flinkService.cancelJobs(flinkInfo.getJobId());
log.info("delete job {} success in backend",
flinkInfo.getJobId());
JobStatus jobStatus =
flinkService.getJobStatus(flinkInfo.getJobId());
if (jobStatus.isTerminalState()) {
- log.info("delete job {} success in backend",
flinkInfo.getJobId());
+ log.info("delete job {} success in backend",
flinkInfo.getJobId());
} else {
- log.info("delete job {} failed in backend",
flinkInfo.getJobId());
+ log.info("delete job {} failed in backend",
flinkInfo.getJobId());
}
} catch (Exception e) {
String msg = String.format("delete job %s failed in
backend exception[%s]", flinkInfo.getJobId(),
diff --git
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/ManagerFlinkTask.java
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/ManagerFlinkTask.java
index 788b64b71..1ddbe1523 100644
---
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/ManagerFlinkTask.java
+++
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/flink/ManagerFlinkTask.java
@@ -57,7 +57,7 @@ public class ManagerFlinkTask {
TaskCommitType.START_NOW.getCode()));
future.get();
} catch (Exception e) {
- log.warn("Flink job some exception [{}]", e.getMessage());
+ log.warn("Flink job some exception [{}]", e);
throw new
BusinessException(BusinessExceptionDesc.UnsupportedOperation
+ e.getMessage());
}
@@ -76,7 +76,7 @@ public class ManagerFlinkTask {
TaskCommitType.RESUME.getCode()));
future.get();
} catch (Exception e) {
- log.warn("Flink job some exception [{}]", e.getMessage());
+ log.warn("Flink job some exception [{}]", e);
throw new
BusinessException(BusinessExceptionDesc.UnsupportedOperation
+ e.getMessage());
}
diff --git
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java
index 3623f7fe9..a6924ff2b 100644
---
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java
+++
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java
@@ -60,9 +60,8 @@ public class DeleteSortListener implements
SortOperateListener {
InlongGroupExtInfo::getKeyValue));
String sortExt = kvConf.get(InlongGroupSettings.SORT_PROPERTIES);
if (StringUtils.isEmpty(sortExt)) {
- String message =
- String.format("inlongGroupId:%s not add deleteProcess
listener,sortProperties is empty",
- inlongGroupId);
+ String message = String.format("groupId [%s] not add deleteProcess
listener, "
+ + "as the sortProperties is empty", inlongGroupId);
log.warn(message);
return ListenerResult.fail(message);
}
@@ -73,7 +72,7 @@ public class DeleteSortListener implements
SortOperateListener {
FlinkInfo flinkInfo = new FlinkInfo();
String jobId = kvConf.get(InlongGroupSettings.SORT_JOB_ID);
- Preconditions.checkNotEmpty(jobId, "sort jobId is empty");
+ Preconditions.checkNotEmpty(jobId, "sortJobId is empty");
flinkInfo.setJobId(jobId);
String sortUrl = kvConf.get(InlongGroupSettings.SORT_URL);
diff --git
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java
index 00e8f6431..47c92547e 100644
---
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java
+++
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java
@@ -72,16 +72,10 @@ public class RestartSortListener implements
SortOperateListener {
Map<String, String> result =
objectMapper.convertValue(objectMapper.readTree(sortExt),
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",
- inlongGroupId);
- log.warn(message);
- return ListenerResult.fail(message);
- }
String dataFlows = kvConf.get(InlongGroupSettings.DATA_FLOW);
if (StringUtils.isEmpty(dataFlows)) {
- String message = String.format("inlongGroupId:{} not add
restartProcess listener,dataflows is empty",
- inlongGroupId);
+ String message = String.format("groupId [%s] not add
restartProcess listener, "
+ + "as the dataflows is empty", inlongGroupId);
log.warn(message);
return ListenerResult.fail(message);
}
@@ -93,8 +87,8 @@ public class RestartSortListener implements
SortOperateListener {
dataFlow = dataflowOptional.get();
}
if (Objects.isNull(dataFlow)) {
- String message = String.format("inlongGroupId:{} not add
restartProcess listener,dataflow is empty",
- inlongGroupId);
+ String message = String.format("groupId [%s] not add
restartProcess listener, "
+ + "as the dataflow is empty", inlongGroupId);
log.warn(message);
return ListenerResult.fail(message);
}
@@ -103,7 +97,7 @@ public class RestartSortListener implements
SortOperateListener {
flinkInfo.setJobName(jobName);
String jobId = kvConf.get(InlongGroupSettings.SORT_JOB_ID);
- Preconditions.checkNotEmpty(jobId, "sort jobId is empty");
+ Preconditions.checkNotEmpty(jobId, "sortJobId is empty");
flinkInfo.setJobId(jobId);
String sortUrl = kvConf.get(InlongGroupSettings.SORT_URL);
diff --git
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
index abca65e64..919379afa 100644
---
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
+++
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
@@ -64,8 +64,8 @@ public class StartupSortListener implements
SortOperateListener {
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",
- inlongGroupId);
+ String message = String.format("groupId [%s] not add
startupProcess listener, "
+ + "as the dataflows is empty", inlongGroupId);
log.warn(message);
return ListenerResult.fail(message);
}
@@ -77,8 +77,8 @@ public class StartupSortListener implements
SortOperateListener {
dataFlow = dataflowOptional.get();
}
if (Objects.isNull(dataFlow)) {
- String message = String.format("inlongGroupId:%s not exec
startupProcess listener,dataflow is empty",
- inlongGroupId);
+ String message = String.format("groupId [%s] not add
startupProcess listener, "
+ + "as the dataflow is empty", inlongGroupId);
log.warn(message);
ListenerResult.fail(message);
}
@@ -106,9 +106,9 @@ public class StartupSortListener implements
SortOperateListener {
try {
managerFlinkTask.start(flinkInfo);
- log.info("the jobId {} submit success");
+ log.info("the jobId {} submit success", flinkInfo.getJobId());
} catch (Exception e) {
- log.warn("startup exception [{}]", e.getMessage());
+ log.warn("startup exception: ", e);
managerFlinkTask.pollFlinkStatus(flinkInfo);
flinkInfo.setException(true);
flinkInfo.setExceptionMsg(getExceptionStackMsg(e));
diff --git
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/SuspendSortListener.java
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/SuspendSortListener.java
index 9c4675549..a642660e1 100644
---
a/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/SuspendSortListener.java
+++
b/inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/SuspendSortListener.java
@@ -59,8 +59,8 @@ public class SuspendSortListener implements
SortOperateListener {
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",
- inlongGroupId);
+ String message = String.format("groupId [%s] not add
suspendProcess listener, "
+ + "as the sortProperties is empty", inlongGroupId);
log.warn(message);
return ListenerResult.fail(message);
}
@@ -71,7 +71,7 @@ public class SuspendSortListener implements
SortOperateListener {
FlinkInfo flinkInfo = new FlinkInfo();
String jobId = kvConf.get(InlongGroupSettings.SORT_JOB_ID);
- Preconditions.checkNotEmpty(jobId, "sort jobId is empty");
+ Preconditions.checkNotEmpty(jobId, "sortJobId is empty");
flinkInfo.setJobId(jobId);
String sortUrl = kvConf.get(InlongGroupSettings.SORT_URL);
diff --git
a/inlong-manager/manager-plugins/src/test/java/org/apache/inlong/manager/plugin/listener/SuspendSortListenerTest.java
b/inlong-manager/manager-plugins/src/test/java/org/apache/inlong/manager/plugin/listener/SuspendSortListenerTest.java
index 32f7f88ef..bbbfd10ea 100644
---
a/inlong-manager/manager-plugins/src/test/java/org/apache/inlong/manager/plugin/listener/SuspendSortListenerTest.java
+++
b/inlong-manager/manager-plugins/src/test/java/org/apache/inlong/manager/plugin/listener/SuspendSortListenerTest.java
@@ -57,7 +57,7 @@ public class SuspendSortListenerTest {
InlongGroupExtInfo inlongGroupExtInfo5 = new InlongGroupExtInfo();
inlongGroupExtInfo5.setKeyName(InlongGroupSettings.SORT_JOB_ID);
- inlongGroupExtInfo5.setKeyValue("118a77173079c7264f08f178da4473fa");
+ inlongGroupExtInfo5.setKeyValue("ea405ab424cfc35ae9be93df8ea87917");
inlongGroupExtInfos.add(inlongGroupExtInfo5);
inlongGroupInfo.setExtList(inlongGroupExtInfos);
