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/24 02:34:30 UTC
[GitHub] [incubator-inlong] luchunliang commented on a change in pull request #2670: [INLONG-2568][DataProxy] Support dynamically getting TubeMq config from Manager
luchunliang commented on a change in pull request #2670:
URL: https://github.com/apache/incubator-inlong/pull/2670#discussion_r813489361
##########
File path: inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/DataProxyClusterServiceImpl.java
##########
@@ -264,16 +279,17 @@ public ProxyPulsarDTO getConfigV2(String dataproxyClusterName) {
*/
Review comment:
It is not suggested to use "/* */ in code block.
##########
File path: inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
##########
@@ -283,21 +291,28 @@ public void diffSetPublish(PulsarClientService pulsarClientService, Set<String>
* @param originalCluster
* @param endCluster
*/
- public void diffRestartPulsarClient(Set<String> originalCluster, Set<String> endCluster) {
- if (!originalCluster.equals(endCluster)) {
- logger.info("pulsarConfig has changed, close current pulsarClientService and restart");
- pulsarClientService.close();
-
- pulsarCluster = configManager.getPulsarUrl2Token();
- configManager.getPulsarConfig().setUrl2token(pulsarCluster);
- pulsarClientService.initCreateConnection(this);
- try {
- initTopicSet(pulsarClientService, new HashSet<String>(topicProperties.values()));
- } catch (Exception e) {
- logger.info("pulsar sink restart, publish topic fail.", e);
- }
+ public void diffUpdatePulsarClient(PulsarClientService pulsarClientService, Map<String, String> originalCluster,
+ Map<String, String> endCluster) {
+ MapDifference<String, String> mapDifference = Maps.difference(originalCluster, endCluster);
+ if (mapDifference.areEqual()) {
+ return;
+ }
+ logger.info("pulsarConfig has changed, close unused url clients and start new url clients");
+ Map<String, String> needToStart = new HashMap<>();
+ Map<String, String> needToClose = new HashMap<>();
+ needToClose.putAll(mapDifference.entriesOnlyOnLeft());
+ needToStart.putAll(mapDifference.entriesOnlyOnRight());
+ Map<String, MapDifference.ValueDifference<String>> differentToken = mapDifference.entriesDiffering();
+ for (String url : differentToken.keySet()) {
+ needToClose.put(url, originalCluster.get(url));
+ needToStart.put(url, endCluster.get(url));//token changed
}
+
+ pulsarClientService.updatePulsarClients(this, needToClose, needToStart,
Review comment:
It is not enough that pulsar client only use two status "start and close".
Closing status is necessary for waiting the completion of sending messages.
##########
File path: inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
##########
@@ -299,6 +300,65 @@ private void destroyConnection() {
logger.debug("closed meta producer");
}
+ /**
+ * close pulsarClients(the related url is removed); start pulsarClients for new url, and create producers for them
+ *
+ * @param callBack
+ * @param needToClose url-token map
+ * @param needToStart url-token map
+ * @param topicSet for new pulsarClient, create these topics' producers
+ */
+ public void updatePulsarClients(CreatePulsarClientCallBack callBack, Map<String, String> needToClose,
+ Map<String, String> needToStart, Set<String> topicSet) {
+ // close
+ for (String url : needToClose.keySet()) {
+ PulsarClient pulsarClient = pulsarClients.get(url);
+ if (pulsarClient != null) {
+ try {
+ pulsarClient.shutdown();
Review comment:
Miss the close operation of topic producer, it will lost the messages in the producer buffer.
##########
File path: inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
##########
@@ -299,6 +300,65 @@ private void destroyConnection() {
logger.debug("closed meta producer");
}
+ /**
+ * close pulsarClients(the related url is removed); start pulsarClients for new url, and create producers for them
+ *
+ * @param callBack
+ * @param needToClose url-token map
+ * @param needToStart url-token map
+ * @param topicSet for new pulsarClient, create these topics' producers
+ */
+ public void updatePulsarClients(CreatePulsarClientCallBack callBack, Map<String, String> needToClose,
+ Map<String, String> needToStart, Set<String> topicSet) {
+ // close
+ for (String url : needToClose.keySet()) {
+ PulsarClient pulsarClient = pulsarClients.get(url);
+ if (pulsarClient != null) {
+ try {
+ pulsarClient.shutdown();
+ pulsarClients.remove(url);
+ } catch (PulsarClientException e) {
+ logger.error("shutdown pulsarClient error in PulsarSink, PulsarClientException {}",
+ e.getMessage());
+ } catch (Exception e) {
+ logger.error("shutdown pulsarClient error in PulsarSink, ex {}", e.getMessage());
+ }
+ }
+ }
+ // new pulsarClient
+ for (Map.Entry<String, String> entry : needToStart.entrySet()) {
+ String url = entry.getKey();
+ String token = entry.getValue();
+ try {
+ if (logger.isDebugEnabled()) {
+ logger.debug("url = {}, token = {}", url, token);
+ }
+ PulsarClient client = initPulsarClient(url, token);
+ pulsarClients.put(url, client);
+ callBack.handleCreateClientSuccess(url);
Review comment:
The operation of "pulsarClients.put(url, client)" must move after the init opeartion of topic producer.
It maybe have crash of the init opeartion of topic producer between sendMessage and updatePulsarClients.
##########
File path: inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
##########
@@ -299,6 +300,65 @@ private void destroyConnection() {
logger.debug("closed meta producer");
}
+ /**
+ * close pulsarClients(the related url is removed); start pulsarClients for new url, and create producers for them
Review comment:
This comment is not fit for the method name "updatePulsarClients"
--
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