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 41f50aa10 [INLONG-4674][Manager] Refactor the client by using the
Retrofit framework (#4681)
41f50aa10 is described below
commit 41f50aa10414d76a600cdfc49c064cd103058693
Author: leosanqing <[email protected]>
AuthorDate: Sun Jun 19 18:44:41 2022 +0800
[INLONG-4674][Manager] Refactor the client by using the Retrofit framework
(#4681)
---
inlong-manager/manager-client/pom.xml | 4 +-
.../client/api/inner/InnerInlongManagerClient.java | 537 +++++++--------------
.../client/api/service/AuthInterceptor.java | 56 +++
.../manager/client/api/service/InlongGroupApi.java | 72 +++
.../client/api/service/InlongStreamApi.java | 55 +++
.../manager/client/api/service/StreamSinkApi.java | 47 ++
.../client/api/service/StreamSourceApi.java | 47 ++
.../client/api/service/StreamTransformApi.java | 48 ++
.../manager/client/api/service/WorkflowApi.java | 44 ++
.../inlong/manager/common/util/JsonUtils.java | 2 +-
licenses/inlong-manager/LICENSE | 2 +
pom.xml | 7 +-
12 files changed, 547 insertions(+), 374 deletions(-)
diff --git a/inlong-manager/manager-client/pom.xml
b/inlong-manager/manager-client/pom.xml
index d6c5e17f3..b4d6f407a 100644
--- a/inlong-manager/manager-client/pom.xml
+++ b/inlong-manager/manager-client/pom.xml
@@ -51,8 +51,8 @@
<artifactId>lombok</artifactId>
</dependency>
<dependency>
- <groupId>com.squareup.okhttp3</groupId>
- <artifactId>okhttp</artifactId>
+ <groupId>com.squareup.retrofit2</groupId>
+ <artifactId>converter-jackson</artifactId>
</dependency>
<dependency>
<groupId>io.swagger</groupId>
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClient.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClient.java
index 6e84472e6..f05d9f00e 100644
---
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClient.java
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/InnerInlongManagerClient.java
@@ -18,22 +18,24 @@
package org.apache.inlong.manager.client.api.inner;
import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.github.pagehelper.PageInfo;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
-import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.Request;
-import okhttp3.Request.Builder;
-import okhttp3.RequestBody;
-import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.inlong.manager.client.api.ClientConfiguration;
import org.apache.inlong.manager.client.api.enums.SimpleGroupStatus;
+import org.apache.inlong.manager.client.api.service.AuthInterceptor;
+import org.apache.inlong.manager.client.api.service.InlongGroupApi;
+import org.apache.inlong.manager.client.api.service.InlongStreamApi;
+import org.apache.inlong.manager.client.api.service.StreamSinkApi;
+import org.apache.inlong.manager.client.api.service.StreamSourceApi;
+import org.apache.inlong.manager.client.api.service.StreamTransformApi;
+import org.apache.inlong.manager.client.api.service.WorkflowApi;
import org.apache.inlong.manager.common.auth.Authentication;
import org.apache.inlong.manager.common.auth.DefaultAuthentication;
import org.apache.inlong.manager.common.beans.Response;
@@ -56,8 +58,13 @@ import
org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
import org.apache.inlong.manager.common.pojo.workflow.form.NewGroupProcessForm;
import org.apache.inlong.manager.common.util.AssertUtils;
import org.apache.inlong.manager.common.util.JsonUtils;
+import retrofit2.Call;
+import retrofit2.Retrofit;
+import retrofit2.converter.jackson.JacksonConverterFactory;
+import java.io.IOException;
import java.util.List;
+import java.util.Map;
/**
* InnerInlongManagerClient is used to invoke http api of inlong manager.
@@ -65,31 +72,49 @@ import java.util.List;
@Slf4j
public class InnerInlongManagerClient {
- protected static final String HTTP_PATH = "api/inlong/manager";
- private static final MediaType APPLICATION_JSON =
MediaType.parse("application/json; charset=utf-8");
- protected final OkHttpClient httpClient;
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
protected final String host;
protected final int port;
- protected final String uname;
- protected final String passwd;
- private final ObjectMapper objectMapper = new ObjectMapper();
+
+ private final InlongStreamApi inlongStreamApi;
+ private final InlongGroupApi inlongGroupApi;
+ private final StreamSourceApi streamSourceApi;
+ private final StreamTransformApi streamTransformApi;
+ private final StreamSinkApi streamSinkApi;
+ private final WorkflowApi workflowApi;
public InnerInlongManagerClient(ClientConfiguration configuration) {
this.host = configuration.getBindHost();
this.port = configuration.getBindPort();
+
Authentication authentication = configuration.getAuthentication();
AssertUtils.notNull(authentication, "Inlong should be authenticated");
AssertUtils.isTrue(authentication instanceof DefaultAuthentication,
"Inlong only support default authentication");
DefaultAuthentication defaultAuthentication = (DefaultAuthentication)
authentication;
- this.uname = defaultAuthentication.getUserName();
- this.passwd = defaultAuthentication.getPassword();
- this.httpClient = new OkHttpClient.Builder()
+
+ OkHttpClient okHttpClient = new OkHttpClient.Builder()
+ .addInterceptor(
+ new
AuthInterceptor(defaultAuthentication.getUserName(),
defaultAuthentication.getPassword()))
.connectTimeout(configuration.getConnectTimeout(),
configuration.getTimeUnit())
.readTimeout(configuration.getReadTimeout(),
configuration.getTimeUnit())
.writeTimeout(configuration.getWriteTimeout(),
configuration.getTimeUnit())
.retryOnConnectionFailure(true)
.build();
+
+ Retrofit retrofit = new Retrofit.Builder()
+ .baseUrl("http://"; + host + ":" + port +
"/api/inlong/manager/")
+
.addConverterFactory(JacksonConverterFactory.create(JsonUtils.OBJECT_MAPPER))
+ .client(okHttpClient)
+ .build();
+
+ inlongStreamApi = retrofit.create(InlongStreamApi.class);
+ inlongGroupApi = retrofit.create(InlongGroupApi.class);
+ streamSinkApi = retrofit.create(StreamSinkApi.class);
+ streamSourceApi = retrofit.create(StreamSourceApi.class);
+ streamTransformApi = retrofit.create(StreamTransformApi.class);
+ workflowApi = retrofit.create(WorkflowApi.class);
}
/**
@@ -111,9 +136,9 @@ public class InnerInlongManagerClient {
public Boolean isGroupExists(String inlongGroupId) {
AssertUtils.notEmpty(inlongGroupId, "InlongGroupId should not be
empty");
- String path = HTTP_PATH + "/group/exist/" + inlongGroupId;
- return this.sendGet(formatUrl(path), new
TypeReference<Response<Boolean>>() {
- });
+ Response<Boolean> response =
executeHttpCall(inlongGroupApi.isGroupExists(inlongGroupId));
+ assertRespSuccess(response);
+ return response.getData();
}
/**
@@ -122,11 +147,7 @@ public class InnerInlongManagerClient {
public InlongGroupInfo getGroupInfo(String inlongGroupId) {
AssertUtils.notEmpty(inlongGroupId, "InlongGroupId should not be
empty");
- String path = HTTP_PATH + "/group/get/" + inlongGroupId;
- final String url = formatUrl(path);
- Response<InlongGroupInfo> responseBody = this.sendGetForResponse(url,
- new TypeReference<Response<InlongGroupInfo>>() {
- });
+ Response<InlongGroupInfo> responseBody =
executeHttpCall(inlongGroupApi.getGroupInfo(inlongGroupId));
if (responseBody.isSuccess()) {
return responseBody.getData();
}
@@ -142,19 +163,15 @@ public class InnerInlongManagerClient {
* Get information of groups.
*/
public PageInfo<InlongGroupListResponse> listGroups(String keyword, int
status, int pageNum, int pageSize) {
- ObjectNode groupQuery = objectMapper.createObjectNode();
- groupQuery.put("keyword", keyword);
- groupQuery.put("status", status);
- groupQuery.put("pageNum", pageNum <= 0 ? 1 : pageNum);
- groupQuery.put("pageSize", pageSize);
-
- String path = HTTP_PATH + "/group/list";
- final String url = formatUrl(path);
- Response<PageInfo<InlongGroupListResponse>> pageInfoResponse =
this.sendPostForResponse(
- url,
- groupQuery.toString(),
- new
TypeReference<Response<PageInfo<InlongGroupListResponse>>>() {
- });
+ InlongGroupPageRequest request = InlongGroupPageRequest.builder()
+ .keyword(keyword)
+ .status(status)
+ .build();
+ request.setPageNum(pageNum <= 0 ? 1 : pageNum);
+ request.setPageSize(pageSize);
+
+ Response<PageInfo<InlongGroupListResponse>> pageInfoResponse =
executeHttpCall(
+ inlongGroupApi.listGroups(request));
if (pageInfoResponse.isSuccess()) {
return pageInfoResponse.getData();
@@ -173,23 +190,19 @@ public class InnerInlongManagerClient {
* @return Response encapsulate of inlong group list
*/
public PageInfo<InlongGroupListResponse> listGroups(InlongGroupPageRequest
pageRequest) {
- return this.sendPost(
- formatUrl(HTTP_PATH + "/group/list"),
- JsonUtils.toJsonString(pageRequest),
- new
TypeReference<Response<PageInfo<InlongGroupListResponse>>>() {
- }
- );
+ Response<PageInfo<InlongGroupListResponse>> pageInfoResponse =
executeHttpCall(
+ inlongGroupApi.listGroups(pageRequest));
+ assertRespSuccess(pageInfoResponse);
+ return pageInfoResponse.getData();
}
/**
* Create inlong group
*/
public String createGroup(InlongGroupRequest groupInfo) {
- return this.sendPost(
- formatUrl(HTTP_PATH + "/group/save"),
- JsonUtils.toJsonString(groupInfo),
- String.class
- );
+ Response<String> response =
executeHttpCall(inlongGroupApi.createGroup(groupInfo));
+ assertRespSuccess(response);
+ return response.getData();
}
/**
@@ -198,24 +211,17 @@ public class InnerInlongManagerClient {
* @return groupId && errMsg
*/
public Pair<String, String> updateGroup(InlongGroupRequest groupRequest) {
- Response<String> updateGroupResp = this.sendPostForResponse(
- formatUrl(HTTP_PATH + "/group/update"),
- JsonUtils.toJsonString(groupRequest),
- String.class
- );
-
- return Pair.of(updateGroupResp.getData(), updateGroupResp.getErrMsg());
+ Response<String> response =
executeHttpCall(inlongGroupApi.updateGroup(groupRequest));
+ return Pair.of(response.getData(), response.getErrMsg());
}
/**
* Create information of stream.
*/
public Integer createStreamInfo(InlongStreamInfo streamInfo) {
- return this.sendPost(
- formatUrl(HTTP_PATH + "/stream/save"),
- JsonUtils.toJsonString(streamInfo),
- Integer.class
- );
+ Response<Integer> response =
executeHttpCall(inlongStreamApi.createStream(streamInfo));
+ assertRespSuccess(response);
+ return response.getData();
}
public Boolean isStreamExists(InlongStreamInfo streamInfo) {
@@ -224,14 +230,13 @@ public class InnerInlongManagerClient {
AssertUtils.notEmpty(groupId, "InlongGroupId should not be empty");
AssertUtils.notEmpty(streamId, "InlongStreamId should not be empty");
- final String url = formatUrl(HTTP_PATH + "/stream/exist/" + groupId +
"/" + streamId);
- return this.sendGet(url, new TypeReference<Response<Boolean>>() {
- });
+ Response<Boolean> response =
executeHttpCall(inlongStreamApi.isStreamExists(groupId, streamId));
+ assertRespSuccess(response);
+ return response.getData();
}
public Pair<Boolean, String> updateStreamInfo(InlongStreamInfo streamInfo)
{
- final String url = formatUrl(HTTP_PATH + "/stream/update");
- Response<Boolean> resp = this.sendPostForResponse(url,
JsonUtils.toJsonString(streamInfo), Boolean.class);
+ Response<Boolean> resp =
executeHttpCall(inlongStreamApi.updateStream(streamInfo));
if (resp.getData() != null) {
return Pair.of(resp.getData(), resp.getErrMsg());
@@ -243,20 +248,16 @@ public class InnerInlongManagerClient {
/**
* Get inlong stream by the given groupId and streamId.
*/
- public InlongStreamInfo getStreamInfo(String inlongGroupId, String
inlongStreamId) {
- String url = formatUrl(HTTP_PATH + "/stream/get");
- url += String.format("&groupId=%s&streamId=%s", inlongGroupId,
inlongStreamId);
- Response<InlongStreamInfo> streamInfoResponse =
this.sendGetForResponse(url,
- new TypeReference<Response<InlongStreamInfo>>() {
- });
+ public InlongStreamInfo getStreamInfo(String groupId, String streamId) {
+ Response<InlongStreamInfo> response =
executeHttpCall(inlongStreamApi.getStream(groupId, streamId));
- if (streamInfoResponse.isSuccess()) {
- return streamInfoResponse.getData();
+ if (response.isSuccess()) {
+ return response.getData();
}
- if (streamInfoResponse.getErrMsg().contains("not exist")) {
+ if (response.getErrMsg().contains("not exist")) {
return null;
} else {
- throw new RuntimeException(streamInfoResponse.getErrMsg());
+ throw new RuntimeException(response.getErrMsg());
}
}
@@ -267,23 +268,18 @@ public class InnerInlongManagerClient {
InlongStreamPageRequest pageRequest = new InlongStreamPageRequest();
pageRequest.setInlongGroupId(inlongGroupId);
- return this.sendPost(
- formatUrl(HTTP_PATH + "/stream/listAll"),
- JsonUtils.toJsonString(pageRequest),
- new TypeReference<Response<PageInfo<FullStreamResponse>>>() {
- }
- ).getList();
+ Response<PageInfo<FullStreamResponse>> response =
executeHttpCall(inlongStreamApi.listStream(pageRequest));
+ assertRespSuccess(response);
+ return response.getData().getList();
}
/**
* Create an inlong stream source.
*/
- public Integer createSource(SourceRequest sourceRequest) {
- return this.sendPost(
- formatUrl(HTTP_PATH + "/source/save"),
- JsonUtils.toJsonString(sourceRequest),
- Integer.class
- );
+ public Integer createSource(SourceRequest request) {
+ Response<Integer> response =
executeHttpCall(streamSourceApi.createSource(request));
+ assertRespSuccess(response);
+ return response.getData();
}
/**
@@ -297,33 +293,21 @@ public class InnerInlongManagerClient {
* List information of sources by the specified source type.
*/
public List<SourceListResponse> listSources(String groupId, String
streamId, String sourceType) {
- String url = formatUrl(HTTP_PATH + "/source/list");
- url = String.format("%s&inlongGroupId=%s&inlongStreamId=%s", url,
groupId, streamId);
- if (StringUtils.isNotEmpty(sourceType)) {
- url = String.format("%s&sourceType=%s", url, sourceType);
- }
-
- return this.sendGet(
- url,
- new TypeReference<Response<PageInfo<SourceListResponse>>>() {
- }
- ).getList();
+ Response<PageInfo<SourceListResponse>> response = executeHttpCall(
+ streamSourceApi.listSources(groupId, streamId, sourceType));
+ assertRespSuccess(response);
+ return response.getData().getList();
}
/**
* Update data Source Information.
*/
- public Pair<Boolean, String> updateSource(SourceRequest sourceRequest) {
- Response<Boolean> resEntity = sendPostForResponse(
- formatUrl(HTTP_PATH + "/source/update"),
- JsonUtils.toJsonString(sourceRequest),
- Boolean.class
- );
-
- if (resEntity.getData() != null) {
- return Pair.of(resEntity.getData(), resEntity.getErrMsg());
+ public Pair<Boolean, String> updateSource(SourceRequest request) {
+ Response<Boolean> response =
executeHttpCall(streamSourceApi.updateSource(request));
+ if (response.getData() != null) {
+ return Pair.of(response.getData(), response.getErrMsg());
} else {
- return Pair.of(false, resEntity.getErrMsg());
+ return Pair.of(false, response.getErrMsg());
}
}
@@ -332,51 +316,40 @@ public class InnerInlongManagerClient {
*/
public boolean deleteSource(int id) {
AssertUtils.isTrue(id > 0, "sourceId is illegal");
- return this.sendDelete(
- formatUrl(HTTP_PATH + "/source/delete/" + id),
- null,
- Boolean.class
- );
+ Response<Boolean> response =
executeHttpCall(streamSourceApi.deleteSource(id));
+ assertRespSuccess(response);
+ return response.getData();
}
/**
* Create a conversion function information.
*/
public Integer createTransform(TransformRequest transformRequest) {
- return this.sendPost(
- formatUrl(HTTP_PATH + "/transform/save"),
- JsonUtils.toJsonString(transformRequest),
- Integer.class
- );
+ Response<Integer> response =
executeHttpCall(streamTransformApi.createTransform(transformRequest));
+ assertRespSuccess(response);
+ return response.getData();
}
/**
* Get all conversion function information.
*/
public List<TransformResponse> listTransform(String groupId, String
streamId) {
- String url = formatUrl(HTTP_PATH + "/transform/list");
- url = String.format("%s&inlongGroupId=%s&inlongStreamId=%s", url,
groupId, streamId);
- return this.sendGet(
- url,
- new TypeReference<Response<List<TransformResponse>>>() {
- }
- );
+ Response<List<TransformResponse>> response = executeHttpCall(
+ streamTransformApi.listTransform(groupId, streamId));
+ assertRespSuccess(response);
+ return response.getData();
}
/**
* Update conversion function information.
*/
public Pair<Boolean, String> updateTransform(TransformRequest
transformRequest) {
- Response<Boolean> responseBody = this.sendPostForResponse(
- formatUrl(HTTP_PATH + "/transform/update"),
- JsonUtils.toJsonString(transformRequest),
- Boolean.class
- );
+ Response<Boolean> response =
executeHttpCall(streamTransformApi.updateTransform(transformRequest));
- if (responseBody.getData() != null) {
- return Pair.of(responseBody.getData(), responseBody.getErrMsg());
+ if (response.getData() != null) {
+ return Pair.of(response.getData(), response.getErrMsg());
} else {
- return Pair.of(false, responseBody.getErrMsg());
+ return Pair.of(false, response.getErrMsg());
}
}
@@ -388,21 +361,17 @@ public class InnerInlongManagerClient {
AssertUtils.notEmpty(transformRequest.getInlongStreamId(),
"inlongStreamId should not be null");
AssertUtils.notEmpty(transformRequest.getTransformName(),
"transformName should not be null");
- String url = formatUrl(HTTP_PATH + "/transform/delete");
- url =
String.format("%s&inlongGroupId=%s&inlongStreamId=%s&transformName=%s", url,
- transformRequest.getInlongGroupId(),
- transformRequest.getInlongStreamId(),
- transformRequest.getTransformName());
-
- return this.sendDelete(url, null, Boolean.class);
+ Response<Boolean> response = executeHttpCall(
+
streamTransformApi.deleteTransform(transformRequest.getInlongGroupId(),
+ transformRequest.getInlongStreamId(),
transformRequest.getTransformName()));
+ assertRespSuccess(response);
+ return response.getData();
}
public Integer createSink(SinkRequest sinkRequest) {
- return this.sendPost(
- formatUrl(HTTP_PATH + "/sink/save"),
- JsonUtils.toJsonString(sinkRequest),
- Integer.class
- );
+ Response<Integer> response =
executeHttpCall(streamSinkApi.createSink(sinkRequest));
+ assertRespSuccess(response);
+ return response.getData();
}
/**
@@ -410,12 +379,9 @@ public class InnerInlongManagerClient {
*/
public boolean deleteSink(int id) {
AssertUtils.isTrue(id > 0, "sinkId is illegal");
-
- return this.sendDelete(
- formatUrl(HTTP_PATH + "/sink/delete/" + id),
- null,
- Boolean.class
- );
+ Response<Boolean> response =
executeHttpCall(streamSinkApi.deleteSink(id));
+ assertRespSuccess(response);
+ return response.getData();
}
/**
@@ -429,28 +395,18 @@ public class InnerInlongManagerClient {
* Get information of data sinks.
*/
public List<SinkListResponse> listSinks(String groupId, String streamId,
String sinkType) {
- String url = formatUrl(HTTP_PATH + "/sink/list");
- url = String.format("%s&inlongGroupId=%s&inlongStreamId=%s", url,
groupId, streamId);
- if (StringUtils.isNotEmpty(sinkType)) {
- url = String.format("%s&sinkType=%s", url, sinkType);
- }
-
- return this.sendGet(
- url,
- new TypeReference<Response<PageInfo<SinkListResponse>>>() {
- }
- ).getList();
+ Response<PageInfo<SinkListResponse>> response = executeHttpCall(
+ streamSinkApi.listSinks(groupId, streamId, sinkType));
+ assertRespSuccess(response);
+ return response.getData().getList();
}
/**
* Update information of data sink.
*/
public Pair<Boolean, String> updateSink(SinkRequest sinkRequest) {
- Response<Boolean> responseBody = this.sendPostForResponse(
- formatUrl(HTTP_PATH + "/sink/update"),
- JsonUtils.toJsonString(sinkRequest),
- Boolean.class
- );
+ Response<Boolean> responseBody =
executeHttpCall(streamSinkApi.updateSink(sinkRequest));
+ assertRespSuccess(responseBody);
if (responseBody.getData() != null) {
return Pair.of(responseBody.getData(), responseBody.getErrMsg());
@@ -460,11 +416,10 @@ public class InnerInlongManagerClient {
}
public WorkflowResult initInlongGroup(InlongGroupRequest groupInfo) {
- return this.sendPost(
- formatUrl(HTTP_PATH + "/group/startProcess/" +
groupInfo.getInlongGroupId()),
- "",
- WorkflowResult.class
- );
+ Response<WorkflowResult> responseBody = executeHttpCall(
+ inlongGroupApi.initInlongGroup(groupInfo.getInlongGroupId()));
+ assertRespSuccess(responseBody);
+ return responseBody.getData();
}
public WorkflowResult startInlongGroup(int taskId, NewGroupProcessForm
newGroupProcessForm) {
@@ -478,14 +433,15 @@ public class InnerInlongManagerClient {
inlongGroupApproveForm.put("formName", "InlongGroupApproveForm");
workflowTaskOperation.set("form", inlongGroupApproveForm);
- String operationData = workflowTaskOperation.toString();
- log.info("startInlongGroup workflowTaskOperation: {}", operationData);
+ log.info("startInlongGroup workflowTaskOperation: {}",
inlongGroupApproveForm);
+
+ Map<String, Object> requestMap =
JsonUtils.OBJECT_MAPPER.convertValue(workflowTaskOperation,
+ new TypeReference<Map<String, Object>>() {
+ });
+ Response<WorkflowResult> response =
executeHttpCall(workflowApi.startInlongGroup(taskId, requestMap));
+ assertRespSuccess(response);
- return this.sendPost(
- formatUrl(HTTP_PATH + "/workflow/approve/" + taskId),
- operationData,
- WorkflowResult.class
- );
+ return response.getData();
}
public boolean operateInlongGroup(String groupId, SimpleGroupStatus
status) {
@@ -493,29 +449,24 @@ public class InnerInlongManagerClient {
}
public boolean operateInlongGroup(String groupId, SimpleGroupStatus
status, boolean async) {
- String path = HTTP_PATH;
+ Call<Response<String>> responseCall;
if (status == SimpleGroupStatus.STOPPED) {
if (async) {
- path += "/group/suspendProcessAsync/";
+ responseCall = inlongGroupApi.suspendProcessAsync(groupId);
} else {
- path += "/group/suspendProcess/";
+ responseCall = inlongGroupApi.suspendProcess(groupId);
}
} else if (status == SimpleGroupStatus.STARTED) {
if (async) {
- path += "/group/restartProcessAsync/";
+ responseCall = inlongGroupApi.restartProcessAsync(groupId);
} else {
- path += "/group/restartProcess/";
+ responseCall = inlongGroupApi.restartProcess(groupId);
}
} else {
throw new IllegalArgumentException(String.format("Unsupported
state: %s", status));
}
- path += groupId;
- Response<String> responseBody = this.sendPostForResponse(
- formatUrl(path),
- null,
- String.class
- );
+ Response<String> responseBody = executeHttpCall(responseCall);
String errMsg = responseBody.getErrMsg();
return responseBody.isSuccess()
@@ -528,22 +479,14 @@ public class InnerInlongManagerClient {
}
public boolean deleteInlongGroup(String groupId, boolean async) {
- String path = HTTP_PATH;
if (async) {
- path += "/group/deleteAsync/" + groupId;
- String finalGroupId = this.sendDelete(
- formatUrl(path),
- null,
- String.class
- );
- return groupId.equals(finalGroupId);
+ Response<String> response =
executeHttpCall(inlongGroupApi.deleteGroupAsync(groupId));
+ assertRespSuccess(response);
+ return groupId.equals(response.getData());
} else {
- path += "/group/delete/" + groupId;
- return this.sendDelete(
- formatUrl(path),
- null,
- Boolean.class
- );
+ Response<Boolean> response =
executeHttpCall(inlongGroupApi.deleteGroup(groupId));
+ assertRespSuccess(response);
+ return response.getData();
}
}
@@ -551,183 +494,37 @@ public class InnerInlongManagerClient {
* get inlong group error messages
*/
public List<EventLogView> getInlongGroupError(String inlongGroupId) {
- String url = formatUrl(HTTP_PATH + "/workflow/event/list");
- url = url + "&inlongGroupId=" + inlongGroupId + "&status=-1";
- return this.sendGet(url, new
TypeReference<Response<PageInfo<EventLogView>>>() {
- }).getList();
+ Response<PageInfo<EventLogView>> response =
executeHttpCall(workflowApi.getInlongGroupError(inlongGroupId, -1));
+ assertRespSuccess(response);
+ return response.getData().getList();
}
/**
* get inlong group error messages
*/
public List<InlongStreamConfigLogListResponse> getStreamLogs(String
inlongGroupId, String inlongStreamId) {
- String url = formatUrl(HTTP_PATH + "/stream/config/log/list");
- url = url + "&inlongGroupId=" + inlongGroupId + "&inlongStreamId=" +
inlongStreamId;
- return this.sendGet(
- url,
- new
TypeReference<Response<PageInfo<InlongStreamConfigLogListResponse>>>() {
- }
- ).getList();
- }
-
- protected String formatUrl(String path) {
- return String.format("http://%s:%s/%s?username=%s&password=%s";, host,
port, path, uname, passwd);
- }
-
- /**
- * Send a GET request, and return the specified type object.
- *
- * @param url request url
- * @param typeReference specified the result type reference
- * @return the result of type T
- */
- private <T> T sendGet(String url, TypeReference<Response<T>>
typeReference) {
- return executeRequestWithCheck("GET", url, null, typeReference);
- }
-
- /**
- * Send a GET request, and return the Response object.
- *
- * @param url request url
- * @param typeReference specified the result type reference
- * @param <T> result type
- * @return the result of type Response<T>
- */
- private <T> Response<T> sendGetForResponse(String url,
TypeReference<Response<T>> typeReference) {
- return executeRequestForResponse("GET", url, null, typeReference);
- }
-
- /**
- * Send a POST request, and return the specified type object.
- *
- * @param url request url
- * @param content request content
- * @param clazz result type
- * @return the result of type T
- */
- private <T> T sendPost(String url, String content, Class<T> clazz) {
- return executeRequestWithCheck("POST", url, content, clazz);
- }
-
- /**
- * Send a POST request, and return the specified type object.
- *
- * @param url request url
- * @param content request content
- * @param typeReference specified the result type reference
- * @return the result of type T
- */
- private <T> T sendPost(String url, String content,
TypeReference<Response<T>> typeReference) {
- return executeRequestWithCheck("POST", url, content, typeReference);
- }
-
- /**
- * Send a POST request, and return the Response object.
- *
- * @param url request url
- * @param content request content
- * @param clazz result type
- * @return the result of type Response<T>
- */
- private <T> Response<T> sendPostForResponse(String url, String content,
Class<T> clazz) {
- return executeRequestForResponse("POST", url, content, clazz);
- }
-
- /**
- * Send a POST request, and return the Response object.
- *
- * @param url request url
- * @param content request content
- * @param typeReference specified the result type reference
- * @return the result of type Response<T>
- */
- private <T> Response<T> sendPostForResponse(String url, String content,
TypeReference<Response<T>> typeReference) {
- return executeRequestForResponse("POST", url, content, typeReference);
- }
-
- /**
- * Send a DELETE request, and return the specified type object.
- *
- * @param url request url
- * @param content request content
- * @param clazz result type
- * @return the result of type T
- */
- private <T> T sendDelete(String url, String content, Class<T> clazz) {
- return executeRequestWithCheck("DELETE", url, content, clazz);
- }
-
- /**
- * Execute the request, and check the status for the response.
- */
- private <T> T executeRequestWithCheck(String method, String url, String
content, Class<T> clazz) {
- Response<T> response = executeRequestForResponse(method, url, content,
clazz);
- Preconditions.checkState(response.isSuccess(), "Inlong request failed:
%s", response.getErrMsg());
- return response.getData();
- }
-
- /**
- * Execute the request, and check the status for the response.
- */
- private <T> T executeRequestWithCheck(String method, String url, String
content,
- TypeReference<Response<T>> typeReference) {
- Response<T> response = executeRequestForResponse(method, url, content,
typeReference);
- Preconditions.checkState(response.isSuccess(), "Inlong request failed:
%s", response.getErrMsg());
- return response.getData();
- }
-
- private <T> Response<T> executeRequestForResponse(String method, String
url, String content, Class<T> clazz) {
- Builder requestBuilder = new Builder().url(url);
- if (content == null) {
- requestBuilder.method(method, null);
- } else {
- requestBuilder.method(method, RequestBody.create(APPLICATION_JSON,
content));
- }
-
- return executeAndParse(requestBuilder.build(), clazz);
- }
-
- private <T> Response<T> executeRequestForResponse(String method, String
url, String content,
- TypeReference<Response<T>> typeReference) {
- Builder requestBuilder = new Builder().url(url);
- if (StringUtils.isBlank(content)) {
- requestBuilder.method(method, null);
- } else {
- requestBuilder.method(method, RequestBody.create(APPLICATION_JSON,
content));
+ Response<PageInfo<InlongStreamConfigLogListResponse>> response =
executeHttpCall(
+ inlongStreamApi.getStreamLogs(inlongGroupId, inlongStreamId));
+ assertRespSuccess(response);
+ return response.getData().getList();
+ }
+
+ private <T> T executeHttpCall(Call<T> call) {
+ Request request = call.request();
+ String url = request.url().encodedPath();
+ try {
+ retrofit2.Response<T> response = call.execute();
+ Preconditions.checkState(response.isSuccessful(),
+ "Request to Inlong %s failed: %s", url,
response.message());
+ return response.body();
+ } catch (IOException e) {
+ log.error(String.format("Request to Inlong %s failed: %s", url,
e.getMessage()), e);
+ throw new RuntimeException(String.format("Request to Inlong %s
failed: %s", url, e.getMessage()), e);
}
-
- return executeAndParse(requestBuilder.build(), typeReference);
}
- private <T> Response<T> executeAndParse(Request request, Class<T> clazz) {
- String body = executeHttpCall(request);
- JavaType javaType =
objectMapper.getTypeFactory().constructParametricType(Response.class, clazz);
- return JsonUtils.parseObject(body, javaType);
- }
-
- private <T> Response<T> executeAndParse(Request request,
TypeReference<Response<T>> typeReference) {
- String body = executeHttpCall(request);
- return JsonUtils.parseObject(body, typeReference);
- }
-
- /**
- * Execute HTTP request call
- *
- * @param request http request
- * @return response body string
- * @throws RuntimeException when response was not success, ex: timeout,
code is not 200
- */
- private String executeHttpCall(Request request) {
- String rul = request.url().encodedPath();
- try (okhttp3.Response response =
httpClient.newCall(request).execute()) {
- assert response.body() != null;
- String body = response.body().string();
- Preconditions.checkState(response.isSuccessful(), "Request to
Inlong %s failed: %s", rul, body);
- return body;
- } catch (Exception e) {
- log.error(String.format("Request to Inlong %s failed: %s", rul,
e.getMessage()), e);
- throw new RuntimeException(String.format("Request to Inlong %s
failed: %s", rul, e.getMessage()), e);
- }
+ private void assertRespSuccess(Response<?> response) {
+ Preconditions.checkState(response.isSuccess(), "Inlong request failed:
%s", response.getErrMsg());
}
}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/AuthInterceptor.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/AuthInterceptor.java
new file mode 100644
index 000000000..822515bc0
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/AuthInterceptor.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.client.api.service;
+
+import okhttp3.HttpUrl;
+import okhttp3.Interceptor;
+import okhttp3.Request;
+import okhttp3.Response;
+
+import java.io.IOException;
+
+/**
+ * AuthInterceptor
+ * Before okhttp call a request, uniformly encapsulate the relevant parameters
of authentication
+ */
+public class AuthInterceptor implements Interceptor {
+
+ private final String username;
+ private final String password;
+
+ public AuthInterceptor(String username, String password) {
+ this.username = username;
+ this.password = password;
+ }
+
+ @Override
+ public Response intercept(Chain chain) throws IOException {
+ Request oldRequest = chain.request();
+ HttpUrl.Builder builder = oldRequest.url()
+ .newBuilder()
+ .addEncodedQueryParameter("username", username)
+ .addEncodedQueryParameter("password", password);
+
+ Request newRequest = oldRequest.newBuilder()
+ .method(oldRequest.method(), oldRequest.body())
+ .url(builder.build())
+ .build();
+
+ return chain.proceed(newRequest);
+ }
+}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InlongGroupApi.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InlongGroupApi.java
new file mode 100644
index 000000000..1cb860764
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InlongGroupApi.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.client.api.service;
+
+import com.github.pagehelper.PageInfo;
+import org.apache.inlong.manager.common.beans.Response;
+import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
+import org.apache.inlong.manager.common.pojo.group.InlongGroupListResponse;
+import org.apache.inlong.manager.common.pojo.group.InlongGroupPageRequest;
+import org.apache.inlong.manager.common.pojo.group.InlongGroupRequest;
+import org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
+import retrofit2.Call;
+import retrofit2.http.Body;
+import retrofit2.http.DELETE;
+import retrofit2.http.GET;
+import retrofit2.http.POST;
+import retrofit2.http.Path;
+
+public interface InlongGroupApi {
+
+ @GET("group/exist/{id}")
+ Call<Response<Boolean>> isGroupExists(@Path("id") String id);
+
+ @GET("group/get/{id}")
+ Call<Response<InlongGroupInfo>> getGroupInfo(@Path("id") String id);
+
+ @POST("group/list")
+ Call<Response<PageInfo<InlongGroupListResponse>>> listGroups(@Body
InlongGroupPageRequest request);
+
+ @POST("group/save")
+ Call<Response<String>> createGroup(@Body InlongGroupRequest request);
+
+ @POST("group/update")
+ Call<Response<String>> updateGroup(@Body InlongGroupRequest request);
+
+ @POST("group/startProcess/{id}")
+ Call<Response<WorkflowResult>> initInlongGroup(@Path("id") String id);
+
+ @POST("group/suspendProcessAsync/{id}")
+ Call<Response<String>> suspendProcessAsync(@Path("id") String id);
+
+ @POST("group/suspendProcess/{id}")
+ Call<Response<String>> suspendProcess(@Path("id") String id);
+
+ @POST("group/restartProcessAsync/{id}")
+ Call<Response<String>> restartProcessAsync(@Path("id") String id);
+
+ @POST("group/restartProcess/{id}")
+ Call<Response<String>> restartProcess(@Path("id") String id);
+
+ @DELETE("group/deleteAsync/{id}")
+ Call<Response<String>> deleteGroupAsync(@Path("id") String id);
+
+ @DELETE("group/delete/{id}")
+ Call<Response<Boolean>> deleteGroup(@Path("id") String id);
+
+}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InlongStreamApi.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InlongStreamApi.java
new file mode 100644
index 000000000..022972c17
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InlongStreamApi.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.client.api.service;
+
+import com.github.pagehelper.PageInfo;
+import org.apache.inlong.manager.common.beans.Response;
+import org.apache.inlong.manager.common.pojo.stream.FullStreamResponse;
+import
org.apache.inlong.manager.common.pojo.stream.InlongStreamConfigLogListResponse;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
+import org.apache.inlong.manager.common.pojo.stream.InlongStreamPageRequest;
+import retrofit2.Call;
+import retrofit2.http.Body;
+import retrofit2.http.GET;
+import retrofit2.http.POST;
+import retrofit2.http.Path;
+import retrofit2.http.Query;
+
+public interface InlongStreamApi {
+
+ @POST("stream/save")
+ Call<Response<Integer>> createStream(@Body InlongStreamInfo stream);
+
+ @GET("stream/exist/{groupId}/{streamId}")
+ Call<Response<Boolean>> isStreamExists(@Path("groupId") String groupId,
@Path("streamId") String streamId);
+
+ @POST("stream/update")
+ Call<Response<Boolean>> updateStream(@Body InlongStreamInfo stream);
+
+ @GET("stream/get")
+ Call<Response<InlongStreamInfo>> getStream(@Query("inlongGroupId") String
groupId,
+ @Query("inlongStreamId") String streamId);
+
+ @POST("stream/listAll")
+ Call<Response<PageInfo<FullStreamResponse>>> listStream(@Body
InlongStreamPageRequest request);
+
+ @GET("stream/config/log/list")
+ Call<Response<PageInfo<InlongStreamConfigLogListResponse>>>
getStreamLogs(@Query("inlongGroupId") String groupId,
+ @Query("inlongStreamId") String streamId);
+
+}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSinkApi.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSinkApi.java
new file mode 100644
index 000000000..47dad5b8d
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSinkApi.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.client.api.service;
+
+import com.github.pagehelper.PageInfo;
+import org.apache.inlong.manager.common.beans.Response;
+import org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
+import org.apache.inlong.manager.common.pojo.sink.SinkRequest;
+import retrofit2.Call;
+import retrofit2.http.Body;
+import retrofit2.http.DELETE;
+import retrofit2.http.GET;
+import retrofit2.http.POST;
+import retrofit2.http.Path;
+import retrofit2.http.Query;
+
+public interface StreamSinkApi {
+
+ @POST("sink/save")
+ Call<Response<Integer>> createSink(@Body SinkRequest request);
+
+ @POST("sink/update")
+ Call<Response<Boolean>> updateSink(@Body SinkRequest request);
+
+ @DELETE("sink/delete/{id}")
+ Call<Response<Boolean>> deleteSink(@Path("id") Integer id);
+
+ @GET("sink/list")
+ Call<Response<PageInfo<SinkListResponse>>>
listSinks(@Query("inlongGroupId") String groupId,
+ @Query("inlongStreamId") String streamId, @Query("sinkType")
String sinkType);
+
+}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSourceApi.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSourceApi.java
new file mode 100644
index 000000000..0c2ba992b
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamSourceApi.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.client.api.service;
+
+import com.github.pagehelper.PageInfo;
+import org.apache.inlong.manager.common.beans.Response;
+import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
+import org.apache.inlong.manager.common.pojo.source.SourceRequest;
+import retrofit2.Call;
+import retrofit2.http.Body;
+import retrofit2.http.DELETE;
+import retrofit2.http.GET;
+import retrofit2.http.POST;
+import retrofit2.http.Path;
+import retrofit2.http.Query;
+
+public interface StreamSourceApi {
+
+ @POST("source/save")
+ Call<Response<Integer>> createSource(@Body SourceRequest request);
+
+ @POST("source/update")
+ Call<Response<Boolean>> updateSource(@Body SourceRequest request);
+
+ @GET("source/list")
+ Call<Response<PageInfo<SourceListResponse>>>
listSources(@Query("inlongGroupId") String groupId,
+ @Query("inlongStreamId") String streamId, @Query("sourceType")
String sourceType);
+
+ @DELETE("source/delete/{id}")
+ Call<Response<Boolean>> deleteSource(@Path("id") Integer sourceId);
+
+}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamTransformApi.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamTransformApi.java
new file mode 100644
index 000000000..ea4221e79
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/StreamTransformApi.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.client.api.service;
+
+import org.apache.inlong.manager.common.beans.Response;
+import org.apache.inlong.manager.common.pojo.transform.TransformRequest;
+import org.apache.inlong.manager.common.pojo.transform.TransformResponse;
+import retrofit2.Call;
+import retrofit2.http.Body;
+import retrofit2.http.DELETE;
+import retrofit2.http.GET;
+import retrofit2.http.POST;
+import retrofit2.http.Path;
+
+import java.util.List;
+
+public interface StreamTransformApi {
+
+ @POST("transform/save")
+ Call<Response<Integer>> createTransform(@Body TransformRequest request);
+
+ @GET("transform/list")
+ Call<Response<List<TransformResponse>>>
listTransform(@Path("inlongGroupId") String groupId,
+ @Path("inlongStreamId") String streamId);
+
+ @POST("transform/update")
+ Call<Response<Boolean>> updateTransform(@Body TransformRequest request);
+
+ @DELETE("transform/delete")
+ Call<Response<Boolean>> deleteTransform(@Path("inlongGroupId") String
groupId,
+ @Path("inlongStreamId") String streamId, @Path("transformName")
String transformName);
+
+}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/WorkflowApi.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/WorkflowApi.java
new file mode 100644
index 000000000..00e816901
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/WorkflowApi.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.client.api.service;
+
+import com.github.pagehelper.PageInfo;
+import org.apache.inlong.manager.common.beans.Response;
+import org.apache.inlong.manager.common.pojo.workflow.EventLogView;
+import org.apache.inlong.manager.common.pojo.workflow.WorkflowResult;
+import retrofit2.Call;
+import retrofit2.http.Body;
+import retrofit2.http.GET;
+import retrofit2.http.Headers;
+import retrofit2.http.POST;
+import retrofit2.http.Path;
+import retrofit2.http.Query;
+
+import java.util.Map;
+
+public interface WorkflowApi {
+
+ @Headers("Content-Type: application/json")
+ @POST("workflow/approve/{taskId}")
+ Call<Response<WorkflowResult>> startInlongGroup(@Path("taskId") Integer
taskId, @Body Map<String, Object> request);
+
+ @GET("workflow/event/list")
+ Call<Response<PageInfo<EventLogView>>>
getInlongGroupError(@Query("inlongGroupId") String groupId,
+ @Query("status") Integer status);
+
+}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/util/JsonUtils.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/util/JsonUtils.java
index 36e527486..de9f22e03 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/util/JsonUtils.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/util/JsonUtils.java
@@ -46,7 +46,7 @@ import java.util.Set;
public class JsonUtils {
public static final String PROJECT_PACKAGE =
"org.apache.inlong.manager.common.pojo";
- private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
static {
OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS,
false);
diff --git a/licenses/inlong-manager/LICENSE b/licenses/inlong-manager/LICENSE
index e2ec3e870..ab517d205 100644
--- a/licenses/inlong-manager/LICENSE
+++ b/licenses/inlong-manager/LICENSE
@@ -671,6 +671,8 @@ The text of each license is the standard Apache 2.0 license.
io.netty:netty-tcnative-classes:2.0.46.Final - Netty/TomcatNative [OpenSSL -
Classes]
(https://github.com/netty/netty-tcnative/tree/netty-tcnative-parent-2.0.46.Final),
(Apache 2.0)
com.nimbusds:nimbus-jose-jwt:7.9 - Nimbus JOSE+JWT
(https://bitbucket.org/connect2id/nimbus-jose-jwt), (The Apache Software
License, Version 2.0)
org.objenesis:objenesis:3.2 - Objenesis (http://objenesis.org/objenesis),
(Apache License, Version 2.0)
+ com.squareup.retrofit2:converter-jackson:2.9.0 - Retrofit
(https://github.com/square/retrofit) (Apache License, Version 2.0)
+ com.squareup.retrofit2:retrofit:2.9.0 - Retrofit
(https://github.com/square/retrofit) (Apache License, Version 2.0)
com.squareup.okhttp:okhttp:2.7.5 - OkHttp
(https://github.com/square/okhttp/tree/parent-2.7.5/okhttp), (Apache 2.0)
com.squareup.okhttp3:okhttp:3.14.9 - OkHttp
(https://github.com/square/okhttp/tree/parent-3.14.9/okhttp), (Apache 2.0)
com.squareup.okio:okio:1.17.2 - Okio
(https://github.com/square/okio/tree/okio-parent-1.17.2/okio), (Apache 2.0)
diff --git a/pom.xml b/pom.xml
index b1dbea47f..9d42d929e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -129,6 +129,7 @@
<httpcore.version>4.4.14</httpcore.version>
<httpclient.version>4.5.13</httpclient.version>
<okhttp.version>3.14.9</okhttp.version>
+ <retrofit.version>2.9.0</retrofit.version>
<spring.boot.version>2.6.6</spring.boot.version>
<spring.version>5.3.20</spring.version>
@@ -815,7 +816,11 @@
<artifactId>okhttp</artifactId>
<version>${okhttp.version}</version>
</dependency>
-
+ <dependency>
+ <groupId>com.squareup.retrofit2</groupId>
+ <artifactId>converter-jackson</artifactId>
+ <version>${retrofit.version}</version>
+ </dependency>
<!-- elastic search -->
<dependency>
<groupId>org.elasticsearch.client</groupId>
