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 11faa02 [INLONG-3136][DataProxy] get "NOUPDATE" configuration from
Manager when request md5 is same (#3138)
11faa02 is described below
commit 11faa02853e51c521296215967ca6f820044310e
Author: 卢春亮 <[email protected]>
AuthorDate: Wed Mar 16 09:27:22 2022 +0800
[INLONG-3136][DataProxy] get "NOUPDATE" configuration from Manager when
request md5 is same (#3138)
---
.../dataproxy/config/RemoteConfigManager.java | 76 ++++++++++++++++++----
.../config/holder/CacheClusterConfigHolder.java | 1 +
.../config/holder/CommonPropertiesHolder.java | 2 +-
.../config/holder/IdTopicConfigHolder.java | 1 +
.../inlong/dataproxy/dispatch/DispatchManager.java | 26 +++++---
.../ManagerPropertiesConfigurationProvider.java | 1 +
.../dataproxy/sink/kafkazone/KafkaZoneSink.java | 1 +
.../sink/kafkazone/KafkaZoneSinkContext.java | 5 +-
.../sink/pulsarzone/PulsarClusterProducer.java | 8 ++-
.../dataproxy/sink/pulsarzone/PulsarZoneSink.java | 1 +
.../sink/pulsarzone/PulsarZoneSinkContext.java | 11 ++--
.../dataproxy/sink/tubezone/TubeZoneSink.java | 1 +
.../sink/tubezone/TubeZoneSinkContext.java | 5 +-
.../main/resources/mappers/ClusterSetMapper.xml | 13 ++--
.../core/impl/DataProxyClusterServiceImpl.java | 14 +++-
.../repository/DataProxyConfigRepository.java | 12 +++-
.../controller/openapi/DataProxyController.java | 2 +-
17 files changed, 133 insertions(+), 47 deletions(-)
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/RemoteConfigManager.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/RemoteConfigManager.java
index f8b5d24..d3dfcf7 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/RemoteConfigManager.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/RemoteConfigManager.java
@@ -78,8 +78,7 @@ public class RemoteConfigManager implements IRepository {
private AtomicInteger managerIpListIndex = new AtomicInteger(0);
// config
private String dataProxyConfigMd5;
- private DataProxyCluster currentClusterConfig;
- private AtomicReference<DataProxyCluster> currentClusterConfigRef;
+ private AtomicReference<DataProxyCluster> currentClusterConfigRef = new
AtomicReference<>();
// flume properties
private Map<String, String> flumeProperties;
// inlong id map
@@ -143,13 +142,13 @@ public class RemoteConfigManager implements IRepository {
//
this.ipListParser.setCommonProperties(ConfigManager.getInstance().getCommonProperties());
List<String> managerIpList = this.ipListParser.getIpList();
+ if (managerIpList == null || managerIpList.size() == 0) {
+ return;
+ }
+ int managerIpSize = managerIpList.size();
for (int i = 0; i < managerIpList.size(); i++) {
- String host =
managerIpList.get(managerIpListIndex.getAndIncrement());
+ String host =
managerIpList.get(managerIpListIndex.getAndIncrement() % managerIpSize);
if (this.reloadDataProxyConfig(proxyClusterName, setName, host)) {
- // parse inlong id
- this.parseInlongIds();
- // generate flume properties
- this.generateFlumeProperties();
break;
}
}
@@ -187,6 +186,7 @@ public class RemoteConfigManager implements IRepository {
// request with get
CloseableHttpResponse response = httpClient.execute(httpGet);
String returnStr = EntityUtils.toString(response.getEntity());
+ LOGGER.info("end to request {} to get config info:{}", url,
returnStr);
// get groupId <-> topic and m value.
DataProxyConfigResponse proxyResponse = gson.fromJson(returnStr,
DataProxyConfigResponse.class);
@@ -194,11 +194,18 @@ public class RemoteConfigManager implements IRepository {
LOGGER.info("Fail to get config info from url:{}, error code
is {}", url, proxyResponse.getErrCode());
return false;
}
+ if (proxyResponse.getErrCode() != DataProxyConfigResponse.SUCC) {
+ LOGGER.info("get config info from url:{}, error code is {}",
url, proxyResponse.getErrCode());
+ return true;
+ }
this.dataProxyConfigMd5 = proxyResponse.getMd5();
DataProxyCluster clusterObj = proxyResponse.getData();
- this.currentClusterConfig = clusterObj;
this.currentClusterConfigRef.set(clusterObj);
+ // parse inlong id
+ this.parseInlongIds();
+ // generate flume properties
+ this.generateFlumeProperties();
} catch (Exception ex) {
LOGGER.error("exception caught", ex);
return false;
@@ -231,8 +238,9 @@ public class RemoteConfigManager implements IRepository {
* @return
*/
public String getZone() {
- if (this.currentClusterConfig != null) {
- return this.currentClusterConfig.getProxyCluster().getZone();
+ DataProxyCluster currentClusterConfig = currentClusterConfigRef.get();
+ if (currentClusterConfig != null) {
+ return currentClusterConfig.getProxyCluster().getZone();
}
return null;
}
@@ -243,8 +251,9 @@ public class RemoteConfigManager implements IRepository {
* @return
*/
public String getProxyClusterName() {
- if (this.currentClusterConfig != null) {
- return this.currentClusterConfig.getProxyCluster().getName();
+ DataProxyCluster currentClusterConfig = currentClusterConfigRef.get();
+ if (currentClusterConfig != null) {
+ return currentClusterConfig.getProxyCluster().getName();
}
return
ConfigManager.getInstance().getCommonProperties().get(KEY_PROXY_CLUSTER_NAME);
}
@@ -255,8 +264,9 @@ public class RemoteConfigManager implements IRepository {
* @return
*/
public String getSetName() {
- if (this.currentClusterConfig != null) {
- return this.currentClusterConfig.getProxyCluster().getSetName();
+ DataProxyCluster currentClusterConfig = currentClusterConfigRef.get();
+ if (currentClusterConfig != null) {
+ return currentClusterConfig.getProxyCluster().getSetName();
}
return
ConfigManager.getInstance().getCommonProperties().get(KEY_SET_NAME);
}
@@ -266,6 +276,7 @@ public class RemoteConfigManager implements IRepository {
*/
private void parseInlongIds() {
Map<String, InLongIdObject> newConfig = new HashMap<>();
+ DataProxyCluster currentClusterConfig = currentClusterConfigRef.get();
ProxyClusterObject proxyClusterObject =
currentClusterConfig.getProxyCluster();
for (InLongIdObject obj : proxyClusterObject.getInlongIds()) {
String inlongId = obj.getInlongId();
@@ -298,6 +309,7 @@ public class RemoteConfigManager implements IRepository {
*/
private void generateFlumeChannels(Map<String, String> newConfig) {
StringBuilder builder = new StringBuilder();
+ DataProxyCluster currentClusterConfig = currentClusterConfigRef.get();
ProxyClusterObject proxyClusterObject =
currentClusterConfig.getProxyCluster();
String proxyClusterName = proxyClusterObject.getName();
// channels
@@ -315,6 +327,17 @@ public class RemoteConfigManager implements IRepository {
newConfig.put(builder.toString(), entry.getValue());
}
}
+ // summary
+ builder.setLength(0);
+ builder.append(proxyClusterName).append(".channels");
+ String key = builder.toString();
+ builder.setLength(0);
+ proxyClusterObject.getChannels().forEach((channel) -> {
+ builder.append(channel.getName()).append(" ");
+ });
+ if (builder.length() > 0) {
+ newConfig.put(key, builder.substring(0, builder.length() - 1));
+ }
}
/**
@@ -325,6 +348,7 @@ public class RemoteConfigManager implements IRepository {
*/
private void generateFlumeSinks(Map<String, String> newConfig) {
StringBuilder builder = new StringBuilder();
+ DataProxyCluster currentClusterConfig = currentClusterConfigRef.get();
ProxyClusterObject proxyClusterObject =
currentClusterConfig.getProxyCluster();
String proxyClusterName = proxyClusterObject.getName();
// sinks
@@ -376,6 +400,17 @@ public class RemoteConfigManager implements IRepository {
}
}
}
+ // summary
+ builder.setLength(0);
+ builder.append(proxyClusterName).append(".sinks");
+ String key = builder.toString();
+ builder.setLength(0);
+ proxyClusterObject.getSinks().forEach((sink) -> {
+ builder.append(sink.getName()).append(" ");
+ });
+ if (builder.length() > 0) {
+ newConfig.put(key, builder.substring(0, builder.length() - 1));
+ }
}
/**
@@ -386,6 +421,7 @@ public class RemoteConfigManager implements IRepository {
*/
private void generateFlumeSources(Map<String, String> newConfig) {
StringBuilder builder = new StringBuilder();
+ DataProxyCluster currentClusterConfig = currentClusterConfigRef.get();
ProxyClusterObject proxyClusterObject =
currentClusterConfig.getProxyCluster();
String proxyClusterName = proxyClusterObject.getName();
// sources
@@ -418,6 +454,17 @@ public class RemoteConfigManager implements IRepository {
newConfig.put(builder.toString(), entry.getValue());
}
}
+ // summary
+ builder.setLength(0);
+ builder.append(proxyClusterName).append(".sources");
+ String key = builder.toString();
+ builder.setLength(0);
+ proxyClusterObject.getSources().forEach((source) -> {
+ builder.append(source.getName()).append(" ");
+ });
+ if (builder.length() > 0) {
+ newConfig.put(key, builder.substring(0, builder.length() - 1));
+ }
}
/**
@@ -445,6 +492,7 @@ public class RemoteConfigManager implements IRepository {
* @return
*/
public DataProxyCluster getCurrentClusterConfig() {
+ DataProxyCluster currentClusterConfig = currentClusterConfigRef.get();
return currentClusterConfig;
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/CacheClusterConfigHolder.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/CacheClusterConfigHolder.java
index a792bfe..01690bd 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/CacheClusterConfigHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/CacheClusterConfigHolder.java
@@ -60,6 +60,7 @@ public class CacheClusterConfigHolder implements Configurable
{
this.reloadInterval = context.getLong(RELOAD_INTERVAL, 60000L);
String loaderType = context.getString(CACHE_CLUSTER_CONFIG_TYPE,
ContextCacheClusterConfigLoader.class.getName());
+ LOG.info("Init CacheClusterConfigLoader,loaderType:{}", loaderType);
try {
Class<?> loaderClass = ClassUtils.getClass(loaderType);
Object loaderObject =
loaderClass.getDeclaredConstructor().newInstance();
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/CommonPropertiesHolder.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/CommonPropertiesHolder.java
index df473b6..ce066fc 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/CommonPropertiesHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/CommonPropertiesHolder.java
@@ -37,7 +37,7 @@ public class CommonPropertiesHolder {
public static final Logger LOG =
LoggerFactory.getLogger(CommonPropertiesHolder.class);
public static final String KEY_COMMON_PROPERTIES =
"common-properties-loader";
public static final String DEFAULT_LOADER =
ClassResourceCommonPropertiesLoader.class.getName();
- public static final String KEY_CLUSTER_ID = "clusterId";
+ public static final String KEY_CLUSTER_ID = "proxy_cluster_name";
private static Map<String, String> props;
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/IdTopicConfigHolder.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/IdTopicConfigHolder.java
index acb653c..9c52361 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/IdTopicConfigHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/IdTopicConfigHolder.java
@@ -63,6 +63,7 @@ public class IdTopicConfigHolder implements Configurable {
this.context = context;
this.reloadInterval = context.getLong(RELOAD_INTERVAL, 60000L);
String loaderType = context.getString(IDTOPIC_CONFIG_TYPE,
ContextIdTopicConfigLoader.class.getName());
+ LOG.info("Init IdTopicConfigLoader,loaderType:{}", loaderType);
try {
Class<?> loaderClass = ClassUtils.getClass(loaderType);
Object loaderObject =
loaderClass.getDeclaredConstructor().newInstance();
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/dispatch/DispatchManager.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/dispatch/DispatchManager.java
index fecae08..57b365e 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/dispatch/DispatchManager.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/dispatch/DispatchManager.java
@@ -23,6 +23,7 @@ import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.flume.Context;
import org.apache.inlong.sdk.commons.protocol.ProxyEvent;
@@ -50,6 +51,8 @@ public class DispatchManager {
private ConcurrentHashMap<String, DispatchProfile> profileCache = new
ConcurrentHashMap<>();
// flag that manager need to output overtime data.
private AtomicBoolean needOutputOvertimeData = new AtomicBoolean(false);
+ private AtomicLong inCounter = new AtomicLong(0);
+ private AtomicLong outCounter = new AtomicLong(0);
/**
* Constructor
@@ -70,10 +73,6 @@ public class DispatchManager {
* @param event
*/
public void addEvent(ProxyEvent event) {
- if (needOutputOvertimeData.get()) {
- this.outputOvertimeData();
- this.needOutputOvertimeData.set(false);
- }
// parse
String eventUid = event.getUid();
long dispatchTime = event.getMsgTime() - event.getMsgTime() %
MINUTE_MS;
@@ -92,8 +91,10 @@ public class DispatchManager {
event.getInlongStreamId(), dispatchTime);
DispatchProfile oldDispatchProfile =
this.profileCache.put(dispatchKey, newDispatchProfile);
this.dispatchQueue.offer(oldDispatchProfile);
+ outCounter.addAndGet(dispatchProfile.getCount());
newDispatchProfile.addEvent(event, maxPackCount, maxPackSize);
}
+ inCounter.incrementAndGet();
}
/**
@@ -102,6 +103,9 @@ public class DispatchManager {
* @return
*/
public void outputOvertimeData() {
+ if (!needOutputOvertimeData.getAndSet(false)) {
+ return;
+ }
LOG.info("start to outputOvertimeData
profileCacheSize:{},dispatchQueueSize:{}",
profileCache.size(), dispatchQueue.size());
long currentTime = System.currentTimeMillis();
@@ -118,10 +122,16 @@ public class DispatchManager {
}
// output
removeKeys.forEach((key) -> {
- dispatchQueue.offer(this.profileCache.remove(key));
+ DispatchProfile dispatchProfile = this.profileCache.remove(key);
+ if (dispatchProfile != null) {
+ dispatchQueue.offer(dispatchProfile);
+ outCounter.addAndGet(dispatchProfile.getCount());
+ }
});
- LOG.info("end to outputOvertimeData
profileCacheSize:{},dispatchQueueSize:{},eventCount:{}",
- profileCache.size(), dispatchQueue.size(), eventCount);
+ LOG.info("end to outputOvertimeData
profileCacheSize:{},dispatchQueueSize:{},eventCount:{},"
+ + "inCounter:{},outCounter:{}",
+ profileCache.size(), dispatchQueue.size(), eventCount,
+ inCounter.getAndSet(0), outCounter.getAndSet(0));
}
/**
@@ -155,6 +165,6 @@ public class DispatchManager {
* setNeedOutputOvertimeData
*/
public void setNeedOutputOvertimeData() {
- this.needOutputOvertimeData.set(true);
+ this.needOutputOvertimeData.getAndSet(true);
}
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/node/ManagerPropertiesConfigurationProvider.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/node/ManagerPropertiesConfigurationProvider.java
index d579abd..d1451ef 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/node/ManagerPropertiesConfigurationProvider.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/node/ManagerPropertiesConfigurationProvider.java
@@ -54,6 +54,7 @@ public class ManagerPropertiesConfigurationProvider extends
public FlumeConfiguration getFlumeConfiguration() {
try {
Map<String, String> flumeProperties =
RemoteConfigManager.getInstance().getFlumeProperties();
+ LOGGER.info("flumeProperties:{}", flumeProperties);
return new FlumeConfiguration(flumeProperties);
} catch (Exception e) {
LOGGER.error("exception catch:" + e.getMessage(), e);
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/kafkazone/KafkaZoneSink.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/kafkazone/KafkaZoneSink.java
index e09f86e..70b8d82 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/kafkazone/KafkaZoneSink.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/kafkazone/KafkaZoneSink.java
@@ -123,6 +123,7 @@ public class KafkaZoneSink extends AbstractSink implements
Configurable {
*/
@Override
public Status process() throws EventDeliveryException {
+ this.dispatchManager.outputOvertimeData();
Channel channel = getChannel();
Transaction tx = channel.getTransaction();
tx.begin();
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/kafkazone/KafkaZoneSinkContext.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/kafkazone/KafkaZoneSinkContext.java
index 4b69e82..a6fc4a4 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/kafkazone/KafkaZoneSinkContext.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/kafkazone/KafkaZoneSinkContext.java
@@ -75,11 +75,12 @@ public class KafkaZoneSinkContext extends SinkContext {
Map<String, String> producerParams =
context.getSubProperties(PREFIX_PRODUCER);
this.producerContext = new Context(producerParams);
// idTopicHolder
+ Context commonPropertiesContext = new
Context(CommonPropertiesHolder.get());
this.idTopicHolder = new IdTopicConfigHolder();
- this.idTopicHolder.configure(context);
+ this.idTopicHolder.configure(commonPropertiesContext);
// cacheHolder
this.cacheHolder = new CacheClusterConfigHolder();
- this.cacheHolder.configure(context);
+ this.cacheHolder.configure(commonPropertiesContext);
}
/**
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarClusterProducer.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarClusterProducer.java
index 0a1e80c..449a92a 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarClusterProducer.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarClusterProducer.java
@@ -129,7 +129,8 @@ public class PulsarClusterProducer implements
LifecycleAware {
.sendTimeout(context.getInteger(KEY_SENDTIMEOUT, 0),
TimeUnit.MILLISECONDS)
.maxPendingMessages(context.getInteger(KEY_MAXPENDINGMESSAGES, 500))
.maxPendingMessagesAcrossPartitions(
-
context.getInteger(KEY_MAXPENDINGMESSAGESACROSSPARTITIONS, 60000))
+
context.getInteger(KEY_MAXPENDINGMESSAGESACROSSPARTITIONS, 60000));
+ this.baseBuilder
.batchingMaxMessages(context.getInteger(KEY_BATCHINGMAXMESSAGES, 500))
.batchingMaxPublishDelay(context.getInteger(KEY_BATCHINGMAXPUBLISHDELAY, 100),
TimeUnit.MILLISECONDS)
@@ -137,7 +138,8 @@ public class PulsarClusterProducer implements
LifecycleAware {
this.baseBuilder
.accessMode(ProducerAccessMode.Shared)
.messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
- .blockIfQueueFull(context.getBoolean(KEY_BLOCKIFQUEUEFULL,
true))
+ .blockIfQueueFull(context.getBoolean(KEY_BLOCKIFQUEUEFULL,
true));
+ this.baseBuilder
.roundRobinRouterBatchingPartitionSwitchFrequency(
context.getInteger(KEY_ROUNDROBINROUTERBATCHINGPARTITIONSWITCHFREQUENCY, 60))
.enableBatching(context.getBoolean(KEY_ENABLEBATCHING,
true))
@@ -153,7 +155,7 @@ public class PulsarClusterProducer implements
LifecycleAware {
* @return CompressionType
*/
private CompressionType getPulsarCompressionType() {
- String type = this.context.getString(KEY_COMPRESSIONTYPE);
+ String type = this.context.getString(KEY_COMPRESSIONTYPE,
CompressionType.SNAPPY.name());
switch (type) {
case "LZ4" :
return CompressionType.LZ4;
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarZoneSink.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarZoneSink.java
index 61aee51..1fa8448 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarZoneSink.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarZoneSink.java
@@ -123,6 +123,7 @@ public class PulsarZoneSink extends AbstractSink implements
Configurable {
*/
@Override
public Status process() throws EventDeliveryException {
+ this.dispatchManager.outputOvertimeData();
Channel channel = getChannel();
Transaction tx = channel.getTransaction();
tx.begin();
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarZoneSinkContext.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarZoneSinkContext.java
index f867f1d..1a1182b 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarZoneSinkContext.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsarzone/PulsarZoneSinkContext.java
@@ -75,11 +75,12 @@ public class PulsarZoneSinkContext extends SinkContext {
Map<String, String> producerParams =
context.getSubProperties(PREFIX_PRODUCER);
this.producerContext = new Context(producerParams);
// idTopicHolder
+ Context commonPropertiesContext = new
Context(CommonPropertiesHolder.get());
this.idTopicHolder = new IdTopicConfigHolder();
- this.idTopicHolder.configure(context);
+ this.idTopicHolder.configure(commonPropertiesContext);
// cacheHolder
this.cacheHolder = new CacheClusterConfigHolder();
- this.cacheHolder.configure(context);
+ this.cacheHolder.configure(commonPropertiesContext);
}
/**
@@ -171,7 +172,7 @@ public class PulsarZoneSinkContext extends SinkContext {
*/
public void addSendMetric(DispatchProfile currentRecord, String bid) {
Map<String, String> dimensions = new HashMap<>();
- dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
this.getClusterId());
+ dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
this.getProxyClusterId());
// metric
fillInlongId(currentRecord, dimensions);
dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getSinkName());
@@ -191,7 +192,7 @@ public class PulsarZoneSinkContext extends SinkContext {
*/
public void addSendFailMetric() {
Map<String, String> dimensions = new HashMap<>();
- dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
this.getClusterId());
+ dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
this.getProxyClusterId());
dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getSinkName());
long msgTime = System.currentTimeMillis();
long auditFormatTime = msgTime - msgTime %
CommonPropertiesHolder.getAuditFormatInterval();
@@ -225,7 +226,7 @@ public class PulsarZoneSinkContext extends SinkContext {
*/
public void addSendResultMetric(DispatchProfile currentRecord, String bid,
boolean result, long sendTime) {
Map<String, String> dimensions = new HashMap<>();
- dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
this.getClusterId());
+ dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
this.getProxyClusterId());
// metric
fillInlongId(currentRecord, dimensions);
dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getSinkName());
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/tubezone/TubeZoneSink.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/tubezone/TubeZoneSink.java
index 6f0ce05..f16148a 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/tubezone/TubeZoneSink.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/tubezone/TubeZoneSink.java
@@ -123,6 +123,7 @@ public class TubeZoneSink extends AbstractSink implements
Configurable {
*/
@Override
public Status process() throws EventDeliveryException {
+ this.dispatchManager.outputOvertimeData();
Channel channel = getChannel();
Transaction tx = channel.getTransaction();
tx.begin();
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/tubezone/TubeZoneSinkContext.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/tubezone/TubeZoneSinkContext.java
index c3e0a22..ceaf2fc 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/tubezone/TubeZoneSinkContext.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/tubezone/TubeZoneSinkContext.java
@@ -75,11 +75,12 @@ public class TubeZoneSinkContext extends SinkContext {
Map<String, String> producerParams =
context.getSubProperties(PREFIX_PRODUCER);
this.producerContext = new Context(producerParams);
// idTopicHolder
+ Context commonPropertiesContext = new
Context(CommonPropertiesHolder.get());
this.idTopicHolder = new IdTopicConfigHolder();
- this.idTopicHolder.configure(context);
+ this.idTopicHolder.configure(commonPropertiesContext);
// cacheHolder
this.cacheHolder = new CacheClusterConfigHolder();
- this.cacheHolder.configure(context);
+ this.cacheHolder.configure(commonPropertiesContext);
}
/**
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/ClusterSetMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/ClusterSetMapper.xml
index f0f9240..8cfeb6e 100644
--- a/inlong-manager/manager-dao/src/main/resources/mappers/ClusterSetMapper.xml
+++ b/inlong-manager/manager-dao/src/main/resources/mappers/ClusterSetMapper.xml
@@ -35,16 +35,13 @@
where is_deleted = 0
</select>
<select id="selectInlongId"
resultType="org.apache.inlong.manager.dao.entity.InLongId">
- select biz.inlong_group_id as inlong_id,
+ select biz.inlong_stream_id as inlong_id,
biz.mq_resource_obj as topic,
- d.sort_type as params,
+ concat('fieldDelimiter=',biz.data_separator) as params,
c.set_name as set_name
- from inlong_group biz,
- data_schema d,
+ from inlong_stream biz,
cluster_set_inlongid c
- where biz.status = 130
- and biz.is_deleted = 0
- and biz.schema_name = d.name
+ where biz.is_deleted = 0
and biz.inlong_group_id = c.inlong_group_id
</select>
<select id="selectCacheCluster"
resultType="org.apache.inlong.manager.dao.entity.CacheCluster">
@@ -89,7 +86,7 @@
where is_deleted = 0
</select>
<select id="selectFlumeSink"
resultType="org.apache.inlong.manager.dao.entity.FlumeSink">
- select channel_name, set_name, type, channel
+ select sink_name, set_name, type, channel
from flume_sink
</select>
<select id="selectFlumeSinkExt"
resultType="org.apache.inlong.manager.dao.entity.FlumeSinkExt">
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/DataProxyClusterServiceImpl.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/DataProxyClusterServiceImpl.java
index 408a66b..69b0147 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/DataProxyClusterServiceImpl.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/DataProxyClusterServiceImpl.java
@@ -19,6 +19,8 @@ package org.apache.inlong.manager.service.core.impl;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
+
+import org.apache.inlong.common.pojo.dataproxy.DataProxyCluster;
import org.apache.inlong.common.pojo.dataproxy.DataProxyConfigResponse;
import org.apache.inlong.manager.common.pojo.dataproxy.DataProxyClusterSet;
import org.apache.inlong.manager.service.core.DataProxyClusterService;
@@ -47,9 +49,19 @@ public class DataProxyClusterServiceImpl implements
DataProxyClusterService {
return this.getErrorAllConfig();
}
String configMd5 = setObj.getMd5Map().get(clusterName);
- if (configMd5 == null || !configMd5.equals(md5)) {
+ if (configMd5 == null) {
return this.getErrorAllConfig();
}
+ // same config
+ if (md5 != null && configMd5.equals(md5)) {
+ DataProxyConfigResponse response = new DataProxyConfigResponse();
+ response.setResult(true);
+ response.setErrCode(DataProxyConfigResponse.NOUPDATE);
+ response.setMd5(configMd5);
+ response.setData(new DataProxyCluster());
+ Gson gson = new Gson();
+ return gson.toJson(response);
+ }
String configJson = setObj.getProxyConfigJson().get(clusterName);
if (configJson == null) {
return this.getErrorAllConfig();
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java
index 215d4d2..7120a12 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java
@@ -121,7 +121,6 @@ public class DataProxyConfigRepository implements
IRepository {
this.reloadCacheCluster(newClusterSets);
this.reloadCacheClusterExt(newClusterSets);
this.reloadCacheTopic(newClusterSets);
- this.reloadProxyCluster(newClusterSets);
this.reloadFlumeChannel(newClusterSets);
this.reloadFlumeChannelExt(newClusterSets);
this.reloadFlumeSource(newClusterSets);
@@ -130,6 +129,7 @@ public class DataProxyConfigRepository implements
IRepository {
this.reloadFlumeSinkExt(newClusterSets);
// reload inlongid
this.reloadInlongId(newClusterSets);
+ this.reloadProxyCluster(newClusterSets);
this.reloadProxy2Cache(newClusterSets);
this.generateClusterJson(newClusterSets);
@@ -238,6 +238,14 @@ public class DataProxyConfigRepository implements
IRepository {
DataProxyClusterSet setObj =
this.getOrCreateDataProxyClusterSet(newClusterSets, setName);
setObj.getProxyClusterList().add(obj);
this.proxyClusterMap.put(obj.getName(), obj);
+ // channels
+ obj.getChannels().addAll(setObj.getProxyChannelMap().values());
+ // inlongids
+ obj.getInlongIds().addAll(setObj.getInlongIds());
+ // sinks
+ obj.getSinks().addAll(setObj.getProxySinkMap().values());
+ // sources
+ obj.getSources().addAll(setObj.getProxySourceMap().values());
}
}
@@ -420,7 +428,7 @@ public class DataProxyConfigRepository implements
IRepository {
response.setErrCode(DataProxyConfigResponse.SUCC);
response.setMd5(md5);
response.setData(clusterObj);
- String jsonResponse = gson.toJson(clusterObj);
+ String jsonResponse = gson.toJson(response);
entry.getValue().getProxyConfigJson().put(proxyObj.getName(),
jsonResponse);
entry.getValue().getMd5Map().put(proxyObj.getName(), md5);
entry.getValue().setDefaultConfigJson(jsonResponse);
diff --git
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/DataProxyController.java
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/DataProxyController.java
index d25d5a3..1f363b1 100644
---
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/DataProxyController.java
+++
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/DataProxyController.java
@@ -71,7 +71,7 @@ public class DataProxyController {
@GetMapping("/getAllConfig")
@ApiOperation(value = "Get all proxy config")
public String getAllConfig(@RequestParam("clusterName") String
clusterName, @RequestParam("setName") String setName,
- @RequestParam("md5") String md5) {
+ @RequestParam(value = "md5", required = false) String md5) {
return dataProxyClusterService.getAllConfig(clusterName, setName, md5);
}
