[INLONG-11585][Manager] Support JDBC verification under dual write pa…

…rameters (#11586)
apache · Dec 8, 2024 · ebcd6d3 · ebcd6d3
1 parent 1183e43
commit ebcd6d3
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 6 deletions.
diff --git a/...anager-pojo/src/main/java/org/apache/inlong/manager/pojo/util/MySQLSensitiveUrlUtils.java b/...anager-pojo/src/main/java/org/apache/inlong/manager/pojo/util/MySQLSensitiveUrlUtils.java
@@ -71,11 +71,13 @@ public static String filterSensitive(String url) {
             }
             resultUrl = resultUrl.replaceAll(InlongConstants.REGEX_WHITESPACE, InlongConstants.EMPTY);
 
-            for (String key : SENSITIVE_REPLACE_PARAM_MAP.keySet()) {
-                resultUrl = StringUtils.replaceIgnoreCase(resultUrl, key + InlongConstants.EQUAL + "true",
+            String sensitiveKey = containSensitiveKey(resultUrl);
+            while (StringUtils.isNotBlank(sensitiveKey)) {
+                resultUrl = StringUtils.replaceIgnoreCase(resultUrl, sensitiveKey + InlongConstants.EQUAL + "true",
                         InlongConstants.EMPTY);
-                resultUrl = StringUtils.replaceIgnoreCase(resultUrl, key + InlongConstants.EQUAL + "yes",
+                resultUrl = StringUtils.replaceIgnoreCase(resultUrl, sensitiveKey + InlongConstants.EQUAL + "yes",
                         InlongConstants.EMPTY);
+                sensitiveKey = containSensitiveKey(resultUrl);
             }
             if (resultUrl.contains(InlongConstants.QUESTION_MARK)) {
                 StringBuilder builder = new StringBuilder();
@@ -114,4 +116,14 @@ public static String filterSensitive(String url) {
                     url, e.getMessage()));
         }
     }
+
+    public static String containSensitiveKey(String url) {
+        for (String key : SENSITIVE_REPLACE_PARAM_MAP.keySet()) {
+            if (url.contains(key + InlongConstants.EQUAL + "true")
+                    || url.contains(key + InlongConstants.EQUAL + "yes")) {
+                return key;
+            }
+        }
+        return null;
+    }
 }
diff --git a/...anager-pojo/src/test/java/org/apache/inlong/manager/pojo/sink/mysql/MySQLSinkDTOTest.java b/...anager-pojo/src/test/java/org/apache/inlong/manager/pojo/sink/mysql/MySQLSinkDTOTest.java
@@ -36,6 +36,10 @@ public void testFilterSensitive() throws Exception {
                 "jdbc:mysql://127.0.0.1:3306?autoReconnect=true&autoDeserialize=false&allowUrlInLocalInfile=false&allowLoadLocalInfile=false",
                 originUrl);
 
+        originUrl = MySQLSinkDTO.filterSensitive(
+                "jdbc:mysql://address=(host=127.0.0.1)(port=3306)(allowLoadallowLoadLocalInfile=trueLocalInfile=true)");
+        Assertions.assertEquals("jdbc:mysql://address=(host=127.0.0.1)(port=3306)()", originUrl);
+
         originUrl = MySQLSinkDTO.filterSensitive(
                 "jdbc:mysql://127.0.0.1:3306?autoReconnect=true&autoDeserialize = TRue&allowLoadLocalInfile=TRue&allowUrlInLocalInfile=TRue&allowLoadLocalInfileInPath=/");
         Assertions.assertEquals(

diff --git a/...ice/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterServiceImpl.java b/...ice/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterServiceImpl.java
@@ -960,7 +960,7 @@ public Boolean testSSHConnection(ClusterNodeRequest request) {
     public DataProxyNodeResponse getDataProxyNodes(String groupId, String protocolType) {
         LOGGER.debug("begin to get data proxy nodes for groupId={}, protocol={}", groupId, protocolType);
 
-        InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
+        InlongGroupEntity groupEntity = groupMapper.selectByGroupIdWithoutTenant(groupId);
         if (groupEntity == null) {
             String errMsg = String.format("group not found by groupId=%s", groupId);
             LOGGER.error(errMsg);
@@ -1082,7 +1082,7 @@ public DataProxyNodeResponse getDataProxyNodesByCluster(String clusterName, Stri
     }
 
     private List<InlongClusterNodeEntity> getClusterNodes(String groupId, String clusterType, String protocolType) {
-        InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
+        InlongGroupEntity groupEntity = groupMapper.selectByGroupIdWithoutTenant(groupId);
         if (groupEntity == null) {
             LOGGER.warn("inlong group not exists for groupId={}", groupId);
             return Lists.newArrayList();

diff --git a/.../org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java b/.../org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
@@ -324,7 +324,7 @@ public List<BriefMQMessage> queryLatestMessages(InlongGroupInfo groupInfo, Inlon
 
         int finalMsgCount = Math.min(request.getMessageCount(), briefMQMessages.size());
         if (finalMsgCount > 0) {
-            return briefMQMessages.subList(0, finalMsgCount);
+            return new ArrayList<>(briefMQMessages.subList(0, finalMsgCount));
         } else {
             return new ArrayList<>();
         }