This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new fa90454bd [INLONG-7666][Manager] Support to freezing and restarting
streamSource (#7667)
fa90454bd is described below
commit fa90454bdb59823cd058311b76fc3550c3bbb58a
Author: fuweng11 <[email protected]>
AuthorDate: Wed Mar 22 20:14:37 2023 +0800
[INLONG-7666][Manager] Support to freezing and restarting streamSource
(#7667)
---
.../service/source/AbstractSourceOperator.java | 6 ++++--
.../service/source/StreamSourceServiceImpl.java | 7 +++++--
.../web/controller/StreamSourceController.java | 16 ++++++++++++++++
.../openapi/OpenStreamSourceController.java | 20 ++++++++++++++++++++
4 files changed, 45 insertions(+), 4 deletions(-)
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
index 704fb30a4..3538515c9 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
@@ -206,7 +206,8 @@ public abstract class AbstractSourceOperator implements
StreamSourceOperator {
return;
}
if (!SourceStatus.isAllowedTransition(curState, nextState)) {
- throw new BusinessException(String.format("source=%s is not
allowed to stop", existEntity));
+ throw new BusinessException(String.format("current source
status=%s for id=%s is not allowed to stop",
+ existEntity.getStatus(), existEntity.getId()));
}
StreamSourceEntity curEntity = CommonBeanUtils.copyProperties(request,
StreamSourceEntity::new);
curEntity.setPreviousStatus(curState.getCode());
@@ -227,7 +228,8 @@ public abstract class AbstractSourceOperator implements
StreamSourceOperator {
SourceStatus curState = SourceStatus.forCode(existEntity.getStatus());
SourceStatus nextState = SourceStatus.TO_BE_ISSUED_ACTIVE;
if (!SourceStatus.isAllowedTransition(curState, nextState)) {
- throw new BusinessException(String.format("Source=%s is not
allowed to restart", existEntity));
+ throw new BusinessException(String.format("current source
status=%s for id=%s is not allowed to restart",
+ existEntity.getStatus(), existEntity.getId()));
}
StreamSourceEntity curEntity = CommonBeanUtils.copyProperties(request,
StreamSourceEntity::new);
curEntity.setPreviousStatus(curState.getCode());
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
index d483bb774..59697c0fc 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
@@ -408,7 +408,9 @@ public class StreamSourceServiceImpl implements
StreamSourceService {
nextStatus = SourceStatus.SOURCE_DISABLE;
}
if (!SourceStatus.isAllowedTransition(curStatus, nextStatus)) {
- throw new BusinessException(String.format("Source=%s is not
allowed to delete", entity));
+ throw new BusinessException(
+ String.format("current source status=%s for id=%s is not
allowed to delete", entity.getStatus(),
+ entity.getId()));
}
entity.setPreviousStatus(curStatus.getCode());
@@ -454,7 +456,8 @@ public class StreamSourceServiceImpl implements
StreamSourceService {
}
if (!SourceStatus.isAllowedTransition(curStatus, nextStatus)) {
throw new BusinessException(ErrorCodeEnum.SOURCE_OPT_NOT_ALLOWED,
- String.format("Source=%s is not allowed to delete",
entity));
+ String.format("current source status=%s for id=%s is not
allowed to delete", entity.getStatus(),
+ entity.getId()));
}
// delete record
entity.setPreviousStatus(curStatus.getCode());
diff --git
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java
index d23354adf..8e8b566eb 100644
---
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java
+++
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java
@@ -88,6 +88,22 @@ public class StreamSourceController {
return Response.success(result);
}
+ @RequestMapping(value = "/source/stop/{id}", method = RequestMethod.POST)
+ @ApiOperation(value = "stop stream source")
+ @ApiImplicitParam(name = "id", dataTypeClass = Integer.class, required =
true)
+ public Response<Boolean> stop(@PathVariable Integer id) {
+ boolean result = sourceService.stop(id,
LoginUserUtils.getLoginUser().getName());
+ return Response.success(result);
+ }
+
+ @RequestMapping(value = "/source/restart/{id}", method =
RequestMethod.POST)
+ @ApiOperation(value = "restart stream source")
+ @ApiImplicitParam(name = "id", dataTypeClass = Integer.class, required =
true)
+ public Response<Boolean> restart(@PathVariable Integer id) {
+ boolean result = sourceService.restart(id,
LoginUserUtils.getLoginUser().getName());
+ return Response.success(result);
+ }
+
@RequestMapping(value = "/source/forceDelete", method =
RequestMethod.DELETE)
@OperationLog(operation = OperationType.DELETE)
@ApiOperation(value = "Force delete stream source by groupId and streamId")
diff --git
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/OpenStreamSourceController.java
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/OpenStreamSourceController.java
index eb27e62e4..a92a8e033 100644
---
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/OpenStreamSourceController.java
+++
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/OpenStreamSourceController.java
@@ -97,4 +97,24 @@ public class OpenStreamSourceController {
Preconditions.expectNotNull(LoginUserUtils.getLoginUser(),
ErrorCodeEnum.LOGIN_USER_EMPTY);
return Response.success(sourceService.delete(id,
LoginUserUtils.getLoginUser()));
}
+
+ @RequestMapping(value = "/source/stop/{id}", method = RequestMethod.POST)
+ @ApiOperation(value = "Stop stream source")
+ @ApiImplicitParam(name = "id", dataTypeClass = Integer.class, required =
true)
+ public Response<Boolean> stop(@PathVariable Integer id) {
+ Preconditions.expectNotNull(id, ErrorCodeEnum.INVALID_PARAMETER,
"sourceId cannot be null");
+ Preconditions.expectNotNull(LoginUserUtils.getLoginUser(),
ErrorCodeEnum.LOGIN_USER_EMPTY);
+ boolean result = sourceService.stop(id,
LoginUserUtils.getLoginUser().getName());
+ return Response.success(result);
+ }
+
+ @RequestMapping(value = "/source/restart/{id}", method =
RequestMethod.POST)
+ @ApiOperation(value = "Restart stream source")
+ @ApiImplicitParam(name = "id", dataTypeClass = Integer.class, required =
true)
+ public Response<Boolean> restart(@PathVariable Integer id) {
+ Preconditions.expectNotNull(id, ErrorCodeEnum.INVALID_PARAMETER,
"sourceId cannot be null");
+ Preconditions.expectNotNull(LoginUserUtils.getLoginUser(),
ErrorCodeEnum.LOGIN_USER_EMPTY);
+ boolean result = sourceService.restart(id,
LoginUserUtils.getLoginUser().getName());
+ return Response.success(result);
+ }
}
