[INLONG-4443][Manager] Remove some deprecated classes (apache#4444)

Author: ciscozhou

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.manager.common.enums;
+package org.apache.inlong.manager.common.pojo.sink.iceberg;
 
 import org.apache.inlong.manager.common.util.Preconditions;
 

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/DBCollectorDetailTaskEntityStatus.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.manager.common.enums;
+package org.apache.inlong.manager.common.pojo.sink.iceberg;
 
 import lombok.Getter;
 
@@ -40,7 +40,7 @@ public enum IcebergType {
     BINARY("binary");
 
     @Getter
-    private String type;
+    private final String type;
 
     IcebergType(String type) {
         this.type = type;

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/DBCollectorTaskConstant.java

@@ -22,7 +22,6 @@
 import com.github.pagehelper.PageInfo;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.inlong.manager.common.beans.ClusterBean;
-import org.apache.inlong.manager.common.enums.Constant;
 import org.apache.inlong.manager.common.enums.ConsumptionStatus;
 import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
 import org.apache.inlong.manager.common.enums.GlobalConstants;
@@ -73,6 +72,10 @@
 @Service
 public class ConsumptionServiceImpl implements ConsumptionService {
 
+    private static final String PREFIX_DLQ = "dlq"; // prefix of the Topic of the dead letter queue
+
+    private static final String PREFIX_RLQ = "rlq"; // prefix of the Topic of the retry letter queue
+
     @Autowired
     private ClusterBean clusterBean;
     @Autowired
@@ -183,7 +186,7 @@ private void savePulsarInfo(ConsumptionMqExtBase mqExtBase, ConsumptionEntity en
         // when closing, delete the related configuration
         String groupId = entity.getInlongGroupId();
         if (dlqEnable) {
-            String dlqTopic = Constant.PREFIX_DLQ + "_" + pulsarInfo.getDeadLetterTopic();
+            String dlqTopic = PREFIX_DLQ + "_" + pulsarInfo.getDeadLetterTopic();
             Boolean exist = streamService.exist(groupId, dlqTopic);
             if (exist) {
                 throw new BusinessException(ErrorCodeEnum.PULSAR_DLQ_DUPLICATED);
@@ -193,7 +196,7 @@ private void savePulsarInfo(ConsumptionMqExtBase mqExtBase, ConsumptionEntity en
             pulsarInfo.setDeadLetterTopic(null);
         }
         if (rlqEnable) {
-            String rlqTopic = Constant.PREFIX_RLQ + "_" + pulsarInfo.getRetryLetterTopic();
+            String rlqTopic = PREFIX_RLQ + "_" + pulsarInfo.getRetryLetterTopic();
             Boolean exist = streamService.exist(groupId, rlqTopic);
             if (exist) {
                 throw new BusinessException(ErrorCodeEnum.PULSAR_RLQ_DUPLICATED);
@@ -267,7 +270,7 @@ public Boolean update(ConsumptionInfo info, String operator) {
                     streamService.logicDeleteDlqOrRlq(groupId, dlqNameOld, operator);
                 } else if (!Objects.equals(dlqNameNew, dlqNameOld)) {
                     pulsarEntity.setIsDlq(1);
-                    String topic = Constant.PREFIX_DLQ + "_" + dlqNameNew;
+                    String topic = PREFIX_DLQ + "_" + dlqNameNew;
                     topic = topic.toLowerCase(Locale.ROOT);
                     pulsarEntity.setDeadLetterTopic(topic);
                     streamService.insertDlqOrRlq(groupId, topic, operator);
@@ -281,7 +284,7 @@ public Boolean update(ConsumptionInfo info, String operator) {
                     streamService.logicDeleteDlqOrRlq(groupId, rlqNameOld, operator);
                 } else if (!Objects.equals(rlqNameNew, pulsarEntity.getRetryLetterTopic())) {
                     pulsarEntity.setIsRlq(1);
-                    String topic = Constant.PREFIX_RLQ + "_" + rlqNameNew;
+                    String topic = PREFIX_RLQ + "_" + rlqNameNew;
                     topic = topic.toLowerCase(Locale.ROOT);
                     pulsarEntity.setRetryLetterTopic(topic);
                     streamService.insertDlqOrRlq(groupId, topic, operator);

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/DBCollectorTaskReturnCode.java

@@ -18,9 +18,6 @@
 package org.apache.inlong.manager.service.core.impl;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.inlong.manager.common.enums.DBCollectorDetailTaskEntityStatus;
-import org.apache.inlong.manager.common.enums.DBCollectorTaskConstant;
-import org.apache.inlong.manager.common.enums.DBCollectorTaskReturnCode;
 import org.apache.inlong.manager.common.pojo.dbcollector.DBCollectorReportTaskRequest;
 import org.apache.inlong.manager.common.pojo.dbcollector.DBCollectorTaskInfo;
 import org.apache.inlong.manager.common.pojo.dbcollector.DBCollectorTaskRequest;
@@ -40,31 +37,34 @@
 public class DBCollectorTaskServiceImpl implements DBCollectorTaskService {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DBCollectorTaskServiceImpl.class);
+    private static final String INTERFACE_VERSION = "1.0";
+
+    private static final int INIT = 0;
+    private static final int DISPATCHED = 1;
+    private static final int DONE = 2;
+    private static final int FAILED = 3;
+
+    private static final int RETURN_SUCC = 0;
+    private static final int RETURN_EMPTY = 1;
+    private static final int RETURN_INVALID_VERSION = 2;
+    private static final int RETURN_INVALID_STATE = 3;
 
     @Autowired
     private DBCollectorDetailTaskMapper detailTaskMapper;
 
     @Override
     public DBCollectorTaskInfo getTask(DBCollectorTaskRequest req) {
         LOGGER.debug("db collector task request: {}", req);
-        if (!DBCollectorTaskConstant.INTERFACE_VERSION.equals(req.getVersion())) {
-            return DBCollectorTaskInfo.builder()
-                    .version(DBCollectorTaskConstant.INTERFACE_VERSION)
-                    .returnCode(DBCollectorTaskReturnCode.INVALID_VERSION.getCode()).build();
+        if (!INTERFACE_VERSION.equals(req.getVersion())) {
+            return DBCollectorTaskInfo.builder().version(INTERFACE_VERSION).returnCode(RETURN_INVALID_VERSION).build();
         }
-        DBCollectorDetailTaskEntity entity = detailTaskMapper
-                .selectOneByState(DBCollectorDetailTaskEntityStatus.INIT.getCode());
+        DBCollectorDetailTaskEntity entity = detailTaskMapper.selectOneByState(INIT);
         if (entity == null) {
-            return DBCollectorTaskInfo.builder()
-                    .version(DBCollectorTaskConstant.INTERFACE_VERSION)
-                    .returnCode(DBCollectorTaskReturnCode.EMPTY.getCode()).build();
+            return DBCollectorTaskInfo.builder().version(INTERFACE_VERSION).returnCode(RETURN_EMPTY).build();
         }
-        int ret = detailTaskMapper.changeState(entity.getId(), 0, DBCollectorDetailTaskEntityStatus.INIT.getCode(),
-                DBCollectorDetailTaskEntityStatus.DISPATCHED.getCode());
+        int ret = detailTaskMapper.changeState(entity.getId(), 0, INIT, DISPATCHED);
         if (ret == 0) {
-            return DBCollectorTaskInfo.builder()
-                    .version(DBCollectorTaskConstant.INTERFACE_VERSION)
-                    .returnCode(DBCollectorTaskReturnCode.EMPTY.getCode()).build();
+            return DBCollectorTaskInfo.builder().version(INTERFACE_VERSION).returnCode(RETURN_EMPTY).build();
         } else {
             DBCollectorTaskInfo.TaskInfo task = new DBCollectorTaskInfo.TaskInfo();
             task.setId(entity.getId());
@@ -82,34 +82,28 @@ public DBCollectorTaskInfo getTask(DBCollectorTaskRequest req) {
             task.setRetryTimes(entity.getRetryTimes());
             task.setInlongGroupId(entity.getGroupId());
             task.setInlongStreamId(entity.getStreamId());
-            return DBCollectorTaskInfo.builder()
-                    .version(DBCollectorTaskConstant.INTERFACE_VERSION)
-                    .returnCode(DBCollectorTaskReturnCode.SUCC.getCode())
-                    .data(task).build();
+            return DBCollectorTaskInfo.builder().version(INTERFACE_VERSION).returnCode(RETURN_SUCC).data(task).build();
         }
     }
 
     @Override
     public Integer reportTask(DBCollectorReportTaskRequest req) {
-        if (!Objects.equals(req.getVersion(), DBCollectorTaskConstant.INTERFACE_VERSION)) {
-            return DBCollectorTaskReturnCode.INVALID_VERSION.getCode();
+        if (!Objects.equals(req.getVersion(), INTERFACE_VERSION)) {
+            return RETURN_INVALID_VERSION;
         }
-        DBCollectorDetailTaskEntity entity = detailTaskMapper
-                .selectByTaskId(req.getId());
+        DBCollectorDetailTaskEntity entity = detailTaskMapper.selectByTaskId(req.getId());
         if (entity == null) {
-            return DBCollectorTaskReturnCode.EMPTY.getCode();
+            return RETURN_EMPTY;
         }
-        if (req.getState() != DBCollectorDetailTaskEntityStatus.DISPATCHED.getCode()
-                && req.getState() != DBCollectorDetailTaskEntityStatus.DONE.getCode()
-                && req.getState() != DBCollectorDetailTaskEntityStatus.FAILED.getCode()) {
-            return DBCollectorTaskReturnCode.INVALID_STATE.getCode();
+        if (req.getState() != DISPATCHED
+                && req.getState() != DONE
+                && req.getState() != FAILED) {
+            return RETURN_INVALID_STATE;
         }
-        int ret = detailTaskMapper
-                .changeState(entity.getId(), req.getOffset(), DBCollectorDetailTaskEntityStatus.DISPATCHED.getCode(),
-                        req.getState());
+        int ret = detailTaskMapper.changeState(entity.getId(), req.getOffset(), DISPATCHED, req.getState());
         if (ret == 0) {
-            return DBCollectorTaskReturnCode.EMPTY.getCode();
+            return RETURN_EMPTY;
         }
-        return DBCollectorTaskReturnCode.SUCC.getCode();
+        return RETURN_SUCC;
     }
 }