This is an automated email from the ASF dual-hosted git repository.
dockerzhang 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 f45e3c51e [INLONG-3591][Sort-Standalone] Support multiple SortTask
with the one-on-one relationship of Source and SortTask. (#3592)
f45e3c51e is described below
commit f45e3c51e9a0716d9e8b91352064029de9eedeff
Author: 卢春亮 <[email protected]>
AuthorDate: Sun Apr 10 15:47:48 2022 +0800
[INLONG-3591][Sort-Standalone] Support multiple SortTask with the
one-on-one relationship of Source and SortTask. (#3592)
---
.../sdk/sort/impl/InLongTopicManagerImpl.java | 4 +-
.../sdk/sort/impl/QueryConsumeConfigImpl.java | 49 ++++----
.../DefaultEvent2IndexRequestHandler.java | 19 ++-
.../standalone/source/sortsdk/SortSdkSource.java | 130 ++++-----------------
4 files changed, 58 insertions(+), 144 deletions(-)
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
index a30f7f44f..250570b1b 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
@@ -66,8 +66,8 @@ public class InLongTopicManagerImpl extends
InLongTopicManager {
super(context, queryConsumeConfig);
updateMetaDataWorker = new
UpdateMetaDataThread(context.getConfig().getUpdateMetaDataIntervalSec(),
TimeUnit.SECONDS);
- String threadName = "sortsdk_inlongtopic_manager_"
- + StringUtil.formatDate(new Date(), "yyyy-MM-dd HH:mm:ss");
+ String threadName = "sortsdk_inlongtopic_manager_" +
context.getConfig().getSortTaskId()
+ + "_" + StringUtil.formatDate(new Date(), "yyyy-MM-dd
HH:mm:ss");
updateMetaDataWorker.start(threadName);
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/QueryConsumeConfigImpl.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/QueryConsumeConfigImpl.java
index 1889e8099..9c7446cc1 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/QueryConsumeConfigImpl.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/QueryConsumeConfigImpl.java
@@ -17,12 +17,12 @@
package org.apache.inlong.sdk.sort.impl;
-import com.google.gson.Gson;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHeaders;
@@ -43,6 +43,9 @@ import org.apache.inlong.sdk.sort.entity.InLongTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.alibaba.fastjson.JSON;
+import com.google.gson.Gson;
+
public class QueryConsumeConfigImpl implements QueryConsumeConfig {
private static final int NOUPDATE_VALUE = 1;
@@ -67,9 +70,9 @@ public class QueryConsumeConfigImpl implements
QueryConsumeConfig {
}
// HTTP GET
- private SortSourceConfigResponse doGetRequest() throws Exception {
+ private SortSourceConfigResponse doGetRequest(String getUrl) throws
Exception {
SortSourceConfigResponse managerResponse;
- HttpGet request = getHttpGet();
+ HttpGet request = getHttpGet(getUrl);
try (CloseableHttpResponse response = httpClient.execute(request)) {
@@ -93,8 +96,8 @@ public class QueryConsumeConfigImpl implements
QueryConsumeConfig {
return null;
}
- private HttpGet getHttpGet() {
- HttpGet request = new HttpGet(getRequestUrlWithParam());
+ private HttpGet getHttpGet(String getUrl) {
+ HttpGet request = new HttpGet(getUrl);
// add request headers
request.addHeader("custom-key", "inlong-readapi");
request.addHeader(HttpHeaders.USER_AGENT, "Googlebot");
@@ -107,12 +110,13 @@ public class QueryConsumeConfigImpl implements
QueryConsumeConfig {
public void reload() {
logger.debug("start to reload sort task config.");
try {
- SortSourceConfigResponse managerResponse = doGetRequest();
+ String getUrl = getRequestUrlWithParam();
+ SortSourceConfigResponse managerResponse = doGetRequest(getUrl);
if (managerResponse == null) {
logger.info("## reload managerResponse == null");
return;
}
- if (handleSortTaskConfResult(managerResponse,
managerResponse.getCode())) {
+ if (handleSortTaskConfResult(getUrl, managerResponse,
managerResponse.getCode())) {
return;
}
} catch (Throwable e) {
@@ -126,34 +130,35 @@ public class QueryConsumeConfigImpl implements
QueryConsumeConfig {
/**
* handle request response
*
- * UPDATE_VALUE = 0; conf update
- * NOUPDATE_VALUE = 1; conf no update, md5 is same
- * REQ_PARAMS_ERROR = -101; request params error
- * FAIL = -1; common error
+ * UPDATE_VALUE = 0; conf update NOUPDATE_VALUE = 1; conf no update, md5
is same REQ_PARAMS_ERROR = -101; request
+ * params error FAIL = -1; common error
*
- * @param response ManagerResponse
- * @param respCodeValue int
- * @return true/false
+ * @param getUrl
+ * @param response ManagerResponse
+ * @param respCodeValue int
+ * @return true/false
*/
- private boolean handleSortTaskConfResult(SortSourceConfigResponse
response, int respCodeValue) throws Exception {
+ private boolean handleSortTaskConfResult(String getUrl,
SortSourceConfigResponse response, int respCodeValue)
+ throws Exception {
switch (respCodeValue) {
- case NOUPDATE_VALUE:
+ case NOUPDATE_VALUE :
logger.debug("manager conf noupdate");
return true;
- case UPDATE_VALUE:
+ case UPDATE_VALUE :
logger.info("manager conf update");
clientContext.getStatManager().getStatistics(clientContext.getConfig().getSortTaskId())
.addManagerConfChangedTimes(1);
this.md5 = response.getMd5();
updateSortTaskConf(response);
break;
- case REQ_PARAMS_ERROR:
+ case REQ_PARAMS_ERROR :
logger.error("return code error:{}", respCodeValue);
clientContext.getStatManager().getStatistics(clientContext.getConfig().getSortTaskId())
.addRequestManagerParamErrorTimes(1);
break;
- default:
- logger.error("return code error:{}", respCodeValue);
+ default :
+ logger.error("return code error:{},request:{},response:{}",
+ respCodeValue, getUrl, JSON.toJSONString(response));
clientContext.getStatManager().getStatistics(clientContext.getConfig().getSortTaskId())
.addRequestManagerCommonErrorTimes(1);
return true;
@@ -186,8 +191,8 @@ public class QueryConsumeConfigImpl implements
QueryConsumeConfig {
/**
* query ConsumeConfig
*
- * @param sortTaskId String
- * @return ConsumeConfig
+ * @param sortTaskId String
+ * @return ConsumeConfig
*/
@Override
public ConsumeConfig queryCurrentConsumeConfig(String sortTaskId) {
diff --git
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/DefaultEvent2IndexRequestHandler.java
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/DefaultEvent2IndexRequestHandler.java
index 64f6385fd..7e0933c67 100644
---
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/DefaultEvent2IndexRequestHandler.java
+++
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/DefaultEvent2IndexRequestHandler.java
@@ -23,9 +23,9 @@ import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.inlong.sdk.commons.protocol.EventConstants;
import org.apache.inlong.sort.standalone.channel.ProfileEvent;
import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
@@ -35,6 +35,8 @@ import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
*/
public class DefaultEvent2IndexRequestHandler implements
IEvent2IndexRequestHandler {
+ public static final String KEY_EXTINFO = "extinfo";
+
private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd
HH:mm:ss");
private AtomicLong esIndexIndex = new
AtomicLong(System.currentTimeMillis());
@@ -110,16 +112,11 @@ public class DefaultEvent2IndexRequestHandler implements
IEvent2IndexRequestHand
* @return
*/
public static String getExtInfo(ProfileEvent event) {
- if (event.getHeaders().size() > 0) {
- StringBuilder sBuilder = new StringBuilder();
- for (Entry<String, String> extInfo :
event.getHeaders().entrySet()) {
- String key = extInfo.getKey();
- String value = extInfo.getValue();
- sBuilder.append(key).append('=').append(value).append('&');
- }
- String extinfo = sBuilder.substring(0, sBuilder.length() - 1);
- return extinfo;
+ String extinfoValue = event.getHeaders().get(KEY_EXTINFO);
+ if (extinfoValue != null) {
+ return KEY_EXTINFO + "=" + extinfoValue;
}
- return "";
+ extinfoValue = KEY_EXTINFO + "=" +
event.getHeaders().get(EventConstants.HEADER_KEY_SOURCE_IP);
+ return extinfoValue;
}
}
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 51ce423d1..d9f0cee2b 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
@@ -19,22 +19,15 @@ package org.apache.inlong.sort.standalone.source.sortsdk;
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.commons.lang.ClassUtils;
import org.apache.flume.Context;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.source.AbstractSource;
-import org.apache.inlong.common.pojo.sortstandalone.SortTaskConfig;
import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
import org.apache.inlong.sdk.sort.api.SortClient;
import org.apache.inlong.sdk.sort.api.SortClientConfig;
@@ -47,11 +40,10 @@ import
org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigType;
import org.apache.inlong.sort.standalone.config.holder.SortSourceConfigType;
import
org.apache.inlong.sort.standalone.config.loader.ClassResourceQueryConsumeConfig;
+import org.apache.inlong.sort.standalone.utils.FlumeConfigGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Preconditions;
-
/**
* Default Source implementation of InLong.
*
@@ -81,8 +73,10 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
// Default consume strategy of {@link SortClient}.
private static final SortClientConfig.ConsumeStrategy defaultStrategy =
SortClientConfig.ConsumeStrategy.lastest;
- // Map of {@link SortClient}.
- private Map<String, SortClient> clients;
+ private String taskName;
+
+ // Context of SortSdkSource.
+ private SortSdkSourceContext context;
// The cluster name of sort.
private String sortClusterName;
@@ -90,18 +84,19 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
// Reload config interval.
private long reloadInterval;
- // Context of SortSdkSource.
- private SortSdkSourceContext context;
-
// Executor for config reloading.
private ScheduledExecutorService pool;
+ // {@link SortClient}.
+ private SortClient sortClient;
+
/**
* Start SortSdkSource.
*/
@Override
public synchronized void start() {
- this.reloadAll();
+ LOG.info("start to SortSdkSource:{}", taskName);
+ this.sortClient = this.newClient(taskName);
}
/**
@@ -110,7 +105,10 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
@Override
public void stop() {
pool.shutdownNow();
- clients.forEach((sortId, client) -> client.close());
+ LOG.info("Close sort client {}.", taskName);
+ if (sortClient != null) {
+ sortClient.close();
+ }
}
/**
@@ -118,7 +116,10 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
*/
@Override
public void run() {
- this.reloadAll();
+ LOG.info("start to reload SortSdkSource:{}", taskName);
+ if (sortClient != null) {
+
sortClient.getConfig().setManagerApiUrl(ManagerUrlHandler.getSortSourceConfigUrl());
+ }
}
/**
@@ -128,10 +129,9 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
*/
@Override
public void configure(Context context) {
- this.clients = new ConcurrentHashMap<>();
- this.sortClusterName =
SortClusterConfigHolder.getClusterConfig().getClusterName();
- Preconditions.checkState(context != null, "No context, configure
failed");
+ this.taskName = context.getString(FlumeConfigGenerator.KEY_TASK_NAME);
this.context = new SortSdkSourceContext(getName(), context);
+ this.sortClusterName =
SortClusterConfigHolder.getClusterConfig().getClusterName();
this.reloadInterval = this.context.getReloadInterval();
this.initReloadExecutor();
@@ -145,94 +145,6 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
pool.scheduleAtFixedRate(this, reloadInterval, reloadInterval,
TimeUnit.SECONDS);
}
- /**
- * Reload clients by current {@link SortTaskConfig}.
- *
- * <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.
- */
- 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 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 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(sortId -> !updatedSortIds.contains(sortId))
- .forEach(sortId -> {
- final SortClient client = clients.get(sortId);
- LOG.info("Close sort client {}.", 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() {
- String configType = CommonPropertiesHolder
- .getString(SortSourceConfigType.KEY_TYPE,
SortSourceConfigType.MANAGER.name());
- if (SortClusterConfigType.MANAGER.name().equalsIgnoreCase(configType))
{
- 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(ManagerUrlHandler.getSortSourceConfigUrl());
- }
-
/**
* Create one {@link SortClient} with specific sort id.
*
@@ -266,7 +178,7 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
new ManagerReportHandlerImpl());
} else if
(SortClusterConfigType.MANAGER.name().equalsIgnoreCase(configType)) {
LOG.info("Create sort sdk client in manager way:{}",
configType);
- this.updateClientConfig(clientConfig);
+
clientConfig.setManagerApiUrl(ManagerUrlHandler.getSortSourceConfigUrl());
client = SortClientFactory.createSortClient(clientConfig);
} else {
LOG.info("Create sort sdk client in custom way:{}",
configType);
