You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2022/02/28 04:07:44 UTC
[GitHub] [incubator-inlong] imvan opened a new pull request #2777: [INLONG-2730][Sort] Reduce the number of AsyncProducerClient
imvan opened a new pull request #2777:
URL: https://github.com/apache/incubator-inlong/pull/2777
### Title Name: [INLONG-2730][Sort] Reduce the number of AsyncProducerClient
Fixes #2730
UT will be done in the last issue #2651
### Motivation
*Explain here the context, and why you're making that change. What is the problem you're trying to solve.*
### Modifications
*Describe the modifications you've done.*
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a followup issue for adding the documentation
--
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
[GitHub] [incubator-inlong] dockerzhang merged pull request #2777: [INLONG-2730][Sort] Reduce the number of AsyncProducerClient
Posted by GitBox <gi...@apache.org>.
dockerzhang merged pull request #2777:
URL: https://github.com/apache/incubator-inlong/pull/2777
--
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
[GitHub] [incubator-inlong] luchunliang commented on a change in pull request #2777: [INLONG-2730][Sort] Reduce the number of AsyncProducerClient
Posted by 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
[GitHub] [incubator-inlong] imvan commented on a change in pull request #2777: [INLONG-2730][Sort] Reduce the number of AsyncProducerClient
Posted by GitBox <gi...@apache.org>.
imvan commented on a change in pull request #2777:
URL: https://github.com/apache/incubator-inlong/pull/2777#discussion_r815614188
##########
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:
have confirmed, it's 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