This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 2a07e11 [INLONG-3237][Sort-Standalone] Support periodiclly update sdk
config and remove expire clients (#3240)
2a07e11 is described below
commit 2a07e112648a0e379834dcdbb9526f229058a291
Author: vernedeng <[email protected]>
AuthorDate: Sat Mar 19 17:20:26 2022 +0800
[INLONG-3237][Sort-Standalone] Support periodiclly update sdk config and
remove expire clients (#3240)
---
.../org/apache/inlong/sdk/sort/api/SortClient.java | 2 +
.../inlong/sdk/sort/impl/SortClientImpl.java | 5 ++
.../standalone/source/sortsdk/SortSdkSource.java | 96 +++++++++++++++++-----
3 files changed, 81 insertions(+), 22 deletions(-)
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClient.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClient.java
index 96bb078..cb3af2f 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClient.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClient.java
@@ -25,4 +25,6 @@ public abstract class SortClient {
throws Exception;
public abstract boolean close();
+
+ public abstract SortClientConfig getConfig();
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java
index 335d610..7ad5857 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java
@@ -123,6 +123,11 @@ public class SortClientImpl extends SortClient {
return (cleanInLongTopicManager && cleanContext);
}
+ @Override
+ public SortClientConfig getConfig() {
+ return this.sortClientConfig;
+ }
+
private InLongTopicFetcher getFetcher(String msgKey) throws
NotExistException {
InLongTopicFetcher inLongTopicFetcher =
inLongTopicManager.getFetcher(msgKey);
if (inLongTopicFetcher == null) {
diff --git
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java
index 9f8d867..1b4eeef 100644
---
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java
+++
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java
@@ -22,10 +22,14 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
import org.apache.flume.Context;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
@@ -83,7 +87,7 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
*/
@Override
public synchronized void start() {
- this.reload();
+ this.reloadAll();
}
/**
@@ -100,7 +104,7 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
*/
@Override
public void run() {
- this.reload();
+ this.reloadAll();
}
/**
@@ -132,30 +136,78 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
* <p> Create new clients with new sort task id, and remove the finished
or scheduled ones. </p>
*
* <p> Current version of SortSdk <b>DO NOT</b> support to get the
corresponding sort id of {@link SortClient}.
- * Hence, the maintenance of mapping of {@literal sortId, SortClient}
should be done by Source itself. Which
- * is not elegant, the <b>REMOVE</b> of expire clients will <b>NOT</b> be
supported right now. </p>
+ * Hence, the maintenance of mapping of {@literal sortId, SortClient}
should be done by Source itself.
*/
- private void reload() {
+ private void reloadAll() {
final List<SortTaskConfig> configs =
SortClusterConfigHolder.getClusterConfig().getSortTasks();
LOG.info("start to reload SortSdkSource");
+ this.startNewClients(configs);
+ this.stopExpiryClients(configs);
+ this.updateAllClientConfig();
+ }
- // Start new clients
- for (SortTaskConfig taskConfig : configs) {
-
- // If exits, skip.
- final String sortId = taskConfig.getName();
- SortClient client = this.clients.get(sortId);
- if (client != null) {
- continue;
- }
-
- // Otherwise, new one client.
- client = this.newClient(sortId);
- if (client != null) {
- this.clients.put(sortId, client);
- }
- }
+ /**
+ * Start a new client from SortTaskConfig.
+ * <p>
+ * If the sortId is in configs, but not in active clients, start it.
+ * </p>
+ *
+ * @param configs Updated SortTaskConfig
+ */
+ private void startNewClients(final List<SortTaskConfig> configs) {
+ configs.stream()
+ .map(SortTaskConfig::getName)
+ .filter(sortId -> !clients.containsKey(sortId))
+ .forEach(sortId -> {
+ final SortClient client = this.newClient(sortId);
+ Optional.ofNullable(client)
+ .ifPresent(c -> clients.put(sortId, c));
+ });
+ }
+
+ /**
+ * Stop an expiry client from SortTaskConfig.
+ * <p>
+ * If the sortId is not in active clients, but not in configs, stop it.
+ * </p>
+ *
+ * @param configs Updated SortTaskConfig
+ */
+ private void stopExpiryClients(final List<SortTaskConfig> configs) {
+ Set<String> updatedSortIds = configs.stream()
+ .map(SortTaskConfig::getName)
+ .collect(Collectors.toSet());
+
+ clients.keySet().stream()
+ .filter(updatedSortIds::contains)
+ .forEach(sortId -> {
+ final SortClient client = clients.get(sortId);
+ try {
+ client.close();
+ } catch (Throwable th) {
+ LOG.error("Got a throwable when close client {}, {}",
sortId, th.getMessage());
+ }
+ clients.remove(sortId);
+ });
+ }
+
+ /**
+ * Update all client config.
+ */
+ private void updateAllClientConfig() {
+ clients.values().stream()
+ .map(SortClient::getConfig)
+ .forEach(this::updateClientConfig);
+ }
+
+ /**
+ * Update one client config.
+ *
+ * @param config The config to be updated.
+ */
+ private void updateClientConfig(SortClientConfig config) {
+
config.setManagerApiUrl(CommonPropertiesHolder.getSourceConfigManagerUrl());
}
/**
@@ -175,7 +227,7 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
SortSdkSource.defaultStrategy,
InetAddress.getLocalHost().getHostAddress());
final FetchCallback callback =
FetchCallback.Factory.create(sortId, getChannelProcessor(), context);
clientConfig.setCallback(callback);
-
clientConfig.setManagerApiUrl(CommonPropertiesHolder.getSourceConfigManagerUrl());
+ this.updateClientConfig(clientConfig);
SortClient client =
SortClientFactory.createSortClient(clientConfig);
client.init();
// temporary use to ACK fetched msg.
