[INLONG-11825][Manager] Support CDC audit ID configuration (#11828)

Author: fuweng11

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/AuditService.java

@@ -71,4 +71,6 @@ public interface AuditService {
      */
     List<AuditProxy> getAuditProxy(String component) throws Exception;
 
+    List<AuditInformation> getCdcAuditInfoList(String type, IndicatorType indicatorType);
+
 }

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java

@@ -67,6 +67,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 /**
  * Audit service layer implementation
@@ -116,6 +117,10 @@ public Boolean refreshBaseItemCache() {
             auditIndicatorMap.clear();
             List<AuditInformation> auditInformationList = AuditOperator.getInstance().getAllAuditInformation();
             List<AuditInformation> metricInformationList = AuditOperator.getInstance().getAllMetricInformation();
+            List<AuditInformation> cdcMetricInformationList = AuditOperator.getInstance().getAllCdcIdInformation();
+            cdcMetricInformationList.forEach(v -> {
+                auditItemMap.put(String.valueOf(v.getAuditId()), v.getNameInChinese());
+            });
             auditInformationList.forEach(v -> {
                 auditItemMap.put(String.valueOf(v.getAuditId()), v.getNameInChinese());
             });
@@ -193,6 +198,15 @@ public List<AuditVO> listByCondition(AuditRequest request) throws Exception {
                 // properly overwrite audit ids by role and stream config
                 if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupEntity.getInlongGroupMode())
                         || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupEntity.getInlongGroupMode())) {
+                    List<AuditInformation> cdcAuditInfoList =
+                            getCdcAuditInfoList(sourceNodeType, IndicatorType.RECEIVED_SUCCESS);
+                    List<String> cdcAuditIdList =
+                            cdcAuditInfoList.stream().map(v -> String.valueOf(v.getAuditId()))
+                                    .collect(Collectors.toList());
+                    if (CollectionUtils.isNotEmpty(cdcAuditIdList)) {
+                        String tempSourceNodeType = sourceNodeType;
+                        cdcAuditIdList.forEach(v -> auditIdMap.put(v, tempSourceNodeType));
+                    }
                     auditIdMap.put(getAuditId(sourceNodeType, IndicatorType.RECEIVED_SUCCESS), sourceNodeType);
                     request.setAuditIds(getAuditIds(groupId, streamId, sourceNodeType, sinkNodeType));
                 } else {
@@ -250,6 +264,14 @@ private List<String> getAuditIds(String groupId, String streamId, String sourceN
             InlongGroupEntity inlongGroup = inlongGroupMapper.selectByGroupId(groupId);
             if (InlongConstants.DATASYNC_REALTIME_MODE.equals(inlongGroup.getInlongGroupMode())
                     || InlongConstants.DATASYNC_OFFLINE_MODE.equals(inlongGroup.getInlongGroupMode())) {
+                List<AuditInformation> cdcAuditInfoList =
+                        getCdcAuditInfoList(sourceNodeType, IndicatorType.RECEIVED_SUCCESS);
+                if (CollectionUtils.isNotEmpty(cdcAuditInfoList)) {
+                    List<String> cdcAuditIdList =
+                            cdcAuditInfoList.stream().map(v -> String.valueOf(v.getAuditId()))
+                                    .collect(Collectors.toList());
+                    auditSet.addAll(cdcAuditIdList);
+                }
                 auditSet.add(getAuditId(sourceNodeType, IndicatorType.RECEIVED_SUCCESS));
             } else {
                 auditSet.add(getAuditId(sinkNodeType, IndicatorType.RECEIVED_SUCCESS));
@@ -294,4 +316,16 @@ public List<AuditProxy> getAuditProxy(String component) throws Exception {
         }
     }
 
+    @Override
+    public List<AuditInformation> getCdcAuditInfoList(String type, IndicatorType indicatorType) {
+        if (StringUtils.isBlank(type)) {
+            return null;
+        }
+
+        FlowType flowType = indicatorType.getCode() % 2 == 0 ? FlowType.INPUT : FlowType.OUTPUT;
+        List<AuditInformation> cdcAuditInfo = AuditOperator.getInstance().getAllCdcIdInformation(type, flowType);
+        Preconditions.expectNotNull(cdcAuditInfo, ErrorCodeEnum.AUDIT_ID_TYPE_NOT_SUPPORTED,
+                String.format(ErrorCodeEnum.AUDIT_ID_TYPE_NOT_SUPPORTED.getMessage(), type));
+        return cdcAuditInfo;
+    }
 }

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sort/SortFlinkConfigOperator.java

@@ -17,11 +17,16 @@
 
 package org.apache.inlong.manager.service.resource.sort;
 
+import org.apache.inlong.audit.CdcIdEnum;
 import org.apache.inlong.audit.entity.AuditComponent;
+import org.apache.inlong.audit.entity.AuditInformation;
 import org.apache.inlong.audit.entity.AuditProxy;
+import org.apache.inlong.audit.entity.CdcType;
+import org.apache.inlong.audit.entity.FlowType;
 import org.apache.inlong.common.enums.IndicatorType;
 import org.apache.inlong.manager.common.consts.InlongConstants;
 import org.apache.inlong.manager.common.consts.SinkType;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
 import org.apache.inlong.manager.common.util.CommonBeanUtils;
 import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
 import org.apache.inlong.manager.pojo.sink.StreamSink;
@@ -302,6 +307,19 @@ private void addAuditId(Map<String, Object> properties, String type, IndicatorTy
             properties.putIfAbsent("metrics.audit.key", auditId);
             properties.putIfAbsent("metrics.audit.proxy.hosts",
                     Joiner.on(InlongConstants.AMPERSAND).join(auditProxyList));
+            List<AuditInformation> cdcAuditInfoList = auditService.getCdcAuditInfoList(type, indicatorType);
+            FlowType flowType = indicatorType.getCode() % 2 == 0 ? FlowType.INPUT : FlowType.OUTPUT;
+            List<String> cdcAuditIdList =
+                    cdcAuditInfoList.stream().map(v -> {
+                        for (CdcType cdcType : CdcType.values()) {
+                            if (Objects.equals(CdcIdEnum.getCdcId(type, flowType, cdcType), v.getAuditId())) {
+                                return cdcType.name() + InlongConstants.EQUAL + v.getAuditId();
+                            }
+                        }
+                        throw new BusinessException("current audit id can not find cdc audit information");
+                    }).collect(Collectors.toList());
+            properties.putIfAbsent("metrics.changelog.audit.key",
+                    Joiner.on(InlongConstants.AMPERSAND).join(cdcAuditIdList));
         } catch (Exception e) {
             LOGGER.error("Current type ={} is not set auditId", type);
         }