[GitHub] [incubator-inlong] luchunliang commented on a change in pull request #2777: [INLONG-2730][Sort] Reduce the number of AsyncProducerClient

[GitBox <gi...@apache.org>]

luchunliang commented on a change in pull request #2777:
URL: https://github.com/apache/incubator-inlong/pull/2777#discussion_r815609768



##########
File path: inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsSinkContext.java
##########
@@ -123,33 +125,44 @@ private void reloadIdParams() {
 
     /**
      * Close expire clients and start new clients.
-     * <p>Each client response for data of one uid.</p>
-     * <p>First, find all UIDs that are in the active clientMap but not in the updated id config (or to say EXPIRE UID),
-     * and put those clients into deletingClientsMap.
+     * 
+     * <p>Each client response for data of one secretId.</p>
+     * <p>First, find all secretId that are in the active clientMap
+     * but not in the updated id config (or to say EXPIRE secretId), and put those clients into deletingClientsMap.
      * The real close process will be done at the beginning of next period of reloading.
-     * Second, find all UIDs that in the updated id config but not in the active clientMap(or to say NEW UID),
-     * and start new clients for these UIDs and put them into the active clientMap.</p>
+     * Second, find all secretIds that in the updated id config
+     * but not in the active clientMap(or to say NEW secretId),
+     * and start new clients for these secretId and put them into the active clientMap.</p>
      */
     private void reloadClients() {
+        // get update secretIds
+        Set<String> updateSecretIdSet = idConfigMap
+                .values()
+                .stream()
+                .map(ClsIdConfig::getSecretId)
+                .collect(Collectors.toSet());
+
+        // remove expire client
         clientMap.keySet()
                 .stream()
-                .filter(uid -> !idConfigMap.containsKey(uid))
+                .filter(secretId -> !updateSecretIdSet.contains(secretId))
                 .forEach(this::removeExpireClient);
-        idConfigMap.keySet()
-                .stream()
-                .filter(uid -> !clientMap.containsKey(uid))
+
+        // start new client
+        updateSecretIdSet.stream()
+                .filter(secretId -> !clientMap.containsKey(secretId))
                 .forEach(this::startNewClient);
     }
 
     /**
      * Start new cls client and put it to the active clientMap.
      *
-     * @param uid UID of new client.
+     * @param secretId SecretId of new client.
      */
-    private void startNewClient(String uid) {
-        ClsIdConfig idConfig = idConfigMap.get(uid);
+    private void startNewClient(String secretId) {
+        ClsIdConfig idConfig = idConfigMap.get(secretId);
         if (idConfig == null) {
-            LOG.error("Start client failed, there is not cls config of {}", uid);
+            LOG.error("Start client failed, there is not cls config of {}", secretId);
             return;
         }
         AsyncProducerConfig producerConfig = new AsyncProducerConfig(

Review comment:
       please check that AsyncProducerClient is thread-safe.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org