Update Runtimer Listener Offer Connect Record (#57)
Update Runtimer Listener Offer Connect Record
diff --git a/adapter/runtimer/README.md b/adapter/runtimer/README.md
index bb66b9e..4e508ba 100644
--- a/adapter/runtimer/README.md
+++ b/adapter/runtimer/README.md
@@ -26,6 +26,17 @@
metrics, logger, exception
detail
+3.10 Note
+本次会议纪要：
+1、runtimer配置从DB加载；
+2、任务配置的命名：targetRunner；
+3、更新任务配置：从Service修改成API+存储 Watch；
+4、EventBusListener： 不能出现RMQ&Message，只能围绕Connect Recorded；
+5、EventBusListener：支持位点提交；
+
+下次会议时间：3月14日，下周二 19:00
+
+
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/Runtimer.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/Runtimer.java
index 6c5095a..896885d 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/Runtimer.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/Runtimer.java
@@ -21,12 +21,10 @@
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.EventRuleTransfer;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.EventTargetPusher;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.ListenerFactory;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.PusherTargetEntity;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetKeyValue;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.RuntimerState;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.ServiceThread;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetKeyValue;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.plugin.Plugin;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfig;
import org.apache.rocketmq.eventbridge.adapter.runtimer.service.PusherConfigManageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,8 +34,6 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventBusListener.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventBusListener.java
index dfeb3e5..0c40235 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventBusListener.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventBusListener.java
@@ -19,7 +19,14 @@
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.internal.DefaultKeyValue;
import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
@@ -28,11 +35,13 @@
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetKeyValue;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.QueueState;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.ServiceThread;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine;
import org.apache.rocketmq.eventbridge.adapter.runtimer.service.PusherConfigManageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
+import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
@@ -116,8 +125,9 @@
return;
}
for (MessageExt messageExt : messageExts) {
- listenerFactory.offerListenEvent(messageExt);
- logger.debug("consumer - {} - offer listen event - {}", JSON.toJSONString(pullConsumer), JSON.toJSON(messageExt));
+ ConnectRecord eventRecord = convertToSinkRecord(messageExt);
+ listenerFactory.offerEventRecord(eventRecord);
+ logger.debug("consumer - {} - offer listen event record - {} - by - message event- {}", JSON.toJSONString(pullConsumer), eventRecord, messageExt);
}
} finally {
pullConsumer.commitSync();
@@ -133,6 +143,36 @@
}
/**
+ * MessageExt convert to connect record
+ * @param messageExt
+ * @return
+ */
+ private ConnectRecord convertToSinkRecord(MessageExt messageExt) {
+ Map<String, String> properties = messageExt.getProperties();
+ Schema schema;
+ Long timestamp;
+ ConnectRecord sinkRecord;
+ String connectTimestamp = properties.get(RuntimerConfigDefine.CONNECT_TIMESTAMP);
+ timestamp = StringUtils.isNotEmpty(connectTimestamp) ? Long.valueOf(connectTimestamp) : null;
+ String connectSchema = properties.get(RuntimerConfigDefine.CONNECT_SCHEMA);
+ schema = StringUtils.isNotEmpty(connectSchema) ? JSON.parseObject(connectSchema, Schema.class) : null;
+ byte[] body = messageExt.getBody();
+ RecordPartition recordPartition = listenerFactory.convertToRecordPartition(messageExt.getTopic(), messageExt.getBrokerName(), messageExt.getQueueId());
+ RecordOffset recordOffset = listenerFactory.convertToRecordOffset(messageExt.getQueueOffset());
+ String bodyStr = new String(body, StandardCharsets.UTF_8);
+ sinkRecord = new ConnectRecord(recordPartition, recordOffset, timestamp, schema, bodyStr);
+ KeyValue keyValue = new DefaultKeyValue();
+ keyValue.put(RuntimerConfigDefine.CONNECT_TOPICNAME, messageExt.getTopic());
+ if (MapUtils.isNotEmpty(properties)) {
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ keyValue.put(entry.getKey(), entry.getValue());
+ }
+ }
+ sinkRecord.addExtension(keyValue);
+ return sinkRecord;
+ }
+
+ /**
* consumer update listener
*/
class ConsumerUpdateListenerImpl implements PusherConfigManageService.TargetConfigUpdateListener {
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventRuleTransfer.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventRuleTransfer.java
index 3be7b8a..f14fdec 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventRuleTransfer.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventRuleTransfer.java
@@ -18,27 +18,17 @@
package org.apache.rocketmq.eventbridge.adapter.runtimer.boot;
import com.alibaba.fastjson.JSON;
-import io.openmessaging.KeyValue;
import io.openmessaging.connector.api.data.ConnectRecord;
-import io.openmessaging.connector.api.data.RecordOffset;
-import io.openmessaging.connector.api.data.RecordPartition;
-import io.openmessaging.connector.api.data.Schema;
-import io.openmessaging.internal.DefaultKeyValue;
-import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.ListenerFactory;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.transfer.TransformEngine;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.PusherTargetEntity;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetKeyValue;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.ServiceThread;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.plugin.Plugin;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimeConfigDefine;
import org.apache.rocketmq.eventbridge.adapter.runtimer.service.PusherConfigManageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
@@ -59,8 +49,6 @@
private ExecutorService executorService = new ThreadPoolExecutor(20,60, 1000,TimeUnit.MICROSECONDS, new LinkedBlockingDeque<>(100));
- private ExecutorService singleExecutor = Executors.newSingleThreadScheduledExecutor();
-
public EventRuleTransfer(Plugin plugin, ListenerFactory listenerFactory, PusherConfigManageService pusherConfigManageService){
this.plugin = plugin;
this.listenerFactory = listenerFactory;
@@ -72,18 +60,6 @@
this.taskTransformMap.putAll(initSinkTaskTransformInfo(taskConfig));
}
- private static final Set<String> MQ_SYS_KEYS = new HashSet<String>() {
- {
- add("MIN_OFFSET");
- add("TRACE_ON");
- add("MAX_OFFSET");
- add("MSG_REGION");
- add("UNIQ_KEY");
- add("WAIT");
- add("TAGS");
- }
- };
-
@Override
public String getServiceName() {
return this.getClass().getSimpleName();
@@ -92,21 +68,20 @@
@Override
public void run() {
while (!stopped){
- MessageExt messageExt = listenerFactory.takeListenerEvent();
- if(Objects.isNull(messageExt)){
- logger.info("listen message is empty, continue by curTime - {}", System.currentTimeMillis());
+ ConnectRecord eventRecord = listenerFactory.takeEventRecord();
+ if(Objects.isNull(eventRecord)){
+ logger.info("listen eventRecord is empty, continue by curTime - {}", System.currentTimeMillis());
this.waitForRunning(1000);
continue;
}
executorService.submit(() -> {
- ConnectRecord connectRecord = convertToSinkDataEntry(messageExt);
// extension add sub
// rule - target
for (TargetKeyValue targetKeyValue : taskTransformMap.keySet()){
// add threadPool for cup task
// attention coreSize
TransformEngine<ConnectRecord> transformEngine = taskTransformMap.get(targetKeyValue);
- ConnectRecord transformRecord = transformEngine.doTransforms(connectRecord);
+ ConnectRecord transformRecord = transformEngine.doTransforms(eventRecord);
if(Objects.isNull(transformRecord)){
continue;
}
@@ -148,41 +123,6 @@
}
/**
- * MessageExt convert to connect record
- * @param message
- * @return
- */
- private ConnectRecord convertToSinkDataEntry(MessageExt message) {
- Map<String, String> properties = message.getProperties();
- Schema schema;
- Long timestamp;
- ConnectRecord sinkDataEntry;
- String connectTimestamp = properties.get(RuntimeConfigDefine.CONNECT_TIMESTAMP);
- timestamp = StringUtils.isNotEmpty(connectTimestamp) ? Long.valueOf(connectTimestamp) : null;
- String connectSchema = properties.get(RuntimeConfigDefine.CONNECT_SCHEMA);
- schema = StringUtils.isNotEmpty(connectSchema) ? JSON.parseObject(connectSchema, Schema.class) : null;
- byte[] body = message.getBody();
- RecordPartition recordPartition = listenerFactory.convertToRecordPartition(message.getTopic(), message.getBrokerName(), message.getQueueId());
- RecordOffset recordOffset = listenerFactory.convertToRecordOffset(message.getQueueOffset());
- String bodyStr = new String(body, StandardCharsets.UTF_8);
- sinkDataEntry = new ConnectRecord(recordPartition, recordOffset, timestamp, schema, bodyStr);
- KeyValue keyValue = new DefaultKeyValue();
- if (MapUtils.isNotEmpty(properties)) {
- for (Map.Entry<String, String> entry : properties.entrySet()) {
- if (MQ_SYS_KEYS.contains(entry.getKey())) {
- keyValue.put("MQ-SYS-" + entry.getKey(), entry.getValue());
- } else if (entry.getKey().startsWith("connect-ext-")) {
- keyValue.put(entry.getKey().replaceAll("connect-ext-", ""), entry.getValue());
- } else {
- keyValue.put(entry.getKey(), entry.getValue());
- }
- }
- }
- sinkDataEntry.addExtension(keyValue);
- return sinkDataEntry;
- }
-
- /**
* transform update listener
*/
class TransformUpdateListenerImpl implements PusherConfigManageService.TargetConfigUpdateListener {
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventTargetPusher.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventTargetPusher.java
index 105c4b6..d324fe4 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventTargetPusher.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/EventTargetPusher.java
@@ -22,7 +22,6 @@
import io.netty.util.internal.ConcurrentSet;
import io.openmessaging.connector.api.component.task.sink.SinkTask;
import io.openmessaging.connector.api.data.ConnectRecord;
-import lombok.SneakyThrows;
import org.apache.commons.collections.MapUtils;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.ListenerFactory;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.pusher.PusherTaskContext;
@@ -31,14 +30,13 @@
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.ServiceThread;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.plugin.Plugin;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.plugin.PluginClassLoader;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimeConfigDefine;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine;
import org.apache.rocketmq.eventbridge.adapter.runtimer.service.PusherConfigManageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.TimeUnit;
/**
* event target push to sink task
@@ -81,7 +79,7 @@
}
for(TargetKeyValue targetKeyValue : taskProperty){
try{
- String taskClass = targetKeyValue.getString(RuntimeConfigDefine.TASK_CLASS);
+ String taskClass = targetKeyValue.getString(RuntimerConfigDefine.TASK_CLASS);
ClassLoader loader = plugin.getPluginClassLoader(taskClass);
Class taskClazz;
boolean isolationFlag = false;
@@ -121,7 +119,7 @@
// task-id for unique-key at ConnectKeyValue
// ConnectKeyValue -> new class for name
// also add in ConnectRecord class system property
- String taskPushName = targetKeyValue.getString(RuntimeConfigDefine.TASK_CLASS);
+ String taskPushName = targetKeyValue.getString(RuntimerConfigDefine.TASK_CLASS);
// add thread pool
for(SinkTask sinkTask : pusherTasks){
if(sinkTask.getClass().getName().equals(taskPushName)){
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/ListenerFactory.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/ListenerFactory.java
index 33804b4..07f010e 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/ListenerFactory.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/ListenerFactory.java
@@ -31,7 +31,7 @@
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.PusherTargetEntity;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetKeyValue;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.LoggerName;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimeConfigDefine;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
@@ -56,6 +56,8 @@
private BlockingQueue<MessageExt> eventMessage = new LinkedBlockingQueue(50000);
+ private BlockingQueue<ConnectRecord> eventRecord = new LinkedBlockingQueue<>(50000);
+
private BlockingQueue<Map<TargetKeyValue, ConnectRecord>> targetQueue = new LinkedBlockingQueue<>(50000);
@Value("${rocketmq.namesrvAddr:}")
@@ -107,6 +109,31 @@
return null;
}
+ /**
+ * offer event record
+ * @param connectRecord
+ * @return
+ */
+ public boolean offerEventRecord(ConnectRecord connectRecord){
+ return eventRecord.offer(connectRecord);
+ }
+
+ /**
+ * take event record
+ * @return
+ */
+ public ConnectRecord takeEventRecord() {
+ if(eventRecord.isEmpty()){
+ return null;
+ }
+ try {
+ return eventRecord.take();
+ }catch (Exception exception){
+ logger.error("take event record exception - stack-> ", exception);
+ }
+ return null;
+ }
+
public String createInstance(String servers) {
String[] serversArray = servers.split(";");
List<String> serversList = new ArrayList<String>();
@@ -125,7 +152,7 @@
* @return
*/
public List<String> parseTopicList(TargetKeyValue taskConfig) {
- String messageQueueStr = taskConfig.getString(RuntimeConfigDefine.CONNECT_TOPICNAME);
+ String messageQueueStr = taskConfig.getString(RuntimerConfigDefine.CONNECT_TOPICNAME);
if (StringUtils.isBlank(messageQueueStr)) {
return null;
}
@@ -141,7 +168,7 @@
public List<String> parseTopicListByList(List<TargetKeyValue> taskConfigs) {
Set<String> allTopicList = Sets.newHashSet();
for(TargetKeyValue taskConfig : taskConfigs){
- String messageQueueStr = taskConfig.getString(RuntimeConfigDefine.CONNECT_TOPICNAME);
+ String messageQueueStr = taskConfig.getString(RuntimerConfigDefine.CONNECT_TOPICNAME);
if (StringUtils.isBlank(messageQueueStr)) {
continue;
}
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/transfer/TransformEngine.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/transfer/TransformEngine.java
index 8c75702..9b59487 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/transfer/TransformEngine.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/transfer/TransformEngine.java
@@ -28,7 +28,7 @@
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.LoggerName;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.plugin.Plugin;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.plugin.PluginClassLoader;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimeConfigDefine;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,7 +49,7 @@
private static final String COMMA = ",";
- private static final String PREFIX = RuntimeConfigDefine.TRANSFORMS + "-";
+ private static final String PREFIX = RuntimerConfigDefine.TRANSFORMS + "-";
public TransformEngine(KeyValue config, Plugin plugin) {
this.config = config;
@@ -59,7 +59,7 @@
}
private void init() {
- String transformsStr = config.getString(RuntimeConfigDefine.TRANSFORMS);
+ String transformsStr = config.getString(RuntimerConfigDefine.TRANSFORMS);
if (StringUtils.isBlank(transformsStr)) {
log.warn("no transforms config, {}", JSON.toJSONString(config));
return;
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimeConfigDefine.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimerConfigDefine.java
similarity index 97%
rename from adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimeConfigDefine.java
rename to adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimerConfigDefine.java
index 3a208d2..ebf1c88 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimeConfigDefine.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimerConfigDefine.java
@@ -24,9 +24,9 @@
import java.util.Set;
/**
- * Define keys for connector and task configs.
+ * Define keys for target runner configs.
*/
-public class RuntimeConfigDefine {
+public class RuntimerConfigDefine {
/**
* The full class name of a specific connector implements.
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/service/PusherConfigManageServiceImpl.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/service/PusherConfigManageServiceImpl.java
index 1dde731..58a66fd 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/service/PusherConfigManageServiceImpl.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/service/PusherConfigManageServiceImpl.java
@@ -20,14 +20,13 @@
import com.google.common.collect.Lists;
import io.openmessaging.KeyValue;
import io.openmessaging.connector.api.component.connector.Connector;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.ListenerFactory;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.FilePathConfigUtil;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.PusherTargetEntity;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetKeyValue;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.plugin.Plugin;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.store.FileBaseKeyValueStore;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.store.KeyValueStore;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimeConfigDefine;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine;
import org.apache.rocketmq.eventbridge.adapter.runtimer.converter.JsonConverter;
import org.apache.rocketmq.eventbridge.adapter.runtimer.converter.ListConverter;
import org.springframework.beans.factory.annotation.Value;
@@ -97,7 +96,7 @@
Map<String, TargetKeyValue> connectorConfigs = connectorKeyValueStore.getKVMap();
for (String connectorName : connectorConfigs.keySet()) {
TargetKeyValue config = connectorConfigs.get(connectorName);
- if (0 != config.getInt(RuntimeConfigDefine.CONFIG_DELETED)) {
+ if (0 != config.getInt(RuntimerConfigDefine.CONFIG_DELETED)) {
continue;
}
result.put(connectorName, config);
@@ -109,9 +108,9 @@
public String putConnectTargetConfig(String connectorName, TargetKeyValue configs) throws Exception {
TargetKeyValue exist = connectorKeyValueStore.get(connectorName);
if (null != exist) {
- Long updateTimestamp = exist.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
+ Long updateTimestamp = exist.getLong(RuntimerConfigDefine.UPDATE_TIMESTAMP);
if (null != updateTimestamp) {
- configs.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, updateTimestamp);
+ configs.put(RuntimerConfigDefine.UPDATE_TIMESTAMP, updateTimestamp);
}
}
if (configs.equals(exist)) {
@@ -119,14 +118,14 @@
}
Long currentTimestamp = System.currentTimeMillis();
- configs.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, currentTimestamp);
- for (String requireConfig : RuntimeConfigDefine.REQUEST_CONFIG) {
+ configs.put(RuntimerConfigDefine.UPDATE_TIMESTAMP, currentTimestamp);
+ for (String requireConfig : RuntimerConfigDefine.REQUEST_CONFIG) {
if (!configs.containsKey(requireConfig)) {
return "Request config key: " + requireConfig;
}
}
- String connectorClass = configs.getString(RuntimeConfigDefine.CONNECTOR_CLASS);
+ String connectorClass = configs.getString(RuntimerConfigDefine.CONNECTOR_CLASS);
ClassLoader classLoader = plugin.getPluginClassLoader(connectorClass);
Class clazz;
if (null != classLoader) {
@@ -144,7 +143,7 @@
@Override
public void recomputeTaskConfigs(String connectorName, Connector connector, Long currentTimestamp, TargetKeyValue configs) {
- int maxTask = configs.getInt(RuntimeConfigDefine.MAX_TASK, 1);
+ int maxTask = configs.getInt(RuntimerConfigDefine.MAX_TASK, 1);
List<KeyValue> taskConfigs = connector.taskConfigs(maxTask);
List<TargetKeyValue> converterdConfigs = new ArrayList<>();
for (KeyValue keyValue : taskConfigs) {
@@ -152,19 +151,19 @@
for (String key : keyValue.keySet()) {
newKeyValue.put(key, keyValue.getString(key));
}
- newKeyValue.put(RuntimeConfigDefine.TASK_CLASS, connector.taskClass().getName());
- newKeyValue.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, currentTimestamp);
+ newKeyValue.put(RuntimerConfigDefine.TASK_CLASS, connector.taskClass().getName());
+ newKeyValue.put(RuntimerConfigDefine.UPDATE_TIMESTAMP, currentTimestamp);
- newKeyValue.put(RuntimeConfigDefine.CONNECT_TOPICNAME, configs.getString(RuntimeConfigDefine.CONNECT_TOPICNAME));
- newKeyValue.put(RuntimeConfigDefine.CONNECT_TOPICNAMES, configs.getString(RuntimeConfigDefine.CONNECT_TOPICNAMES));
+ newKeyValue.put(RuntimerConfigDefine.CONNECT_TOPICNAME, configs.getString(RuntimerConfigDefine.CONNECT_TOPICNAME));
+ newKeyValue.put(RuntimerConfigDefine.CONNECT_TOPICNAMES, configs.getString(RuntimerConfigDefine.CONNECT_TOPICNAMES));
Set<String> connectConfigKeySet = configs.keySet();
for (String connectConfigKey : connectConfigKeySet) {
- if (connectConfigKey.startsWith(RuntimeConfigDefine.TRANSFORMS)) {
+ if (connectConfigKey.startsWith(RuntimerConfigDefine.TRANSFORMS)) {
newKeyValue.put(connectConfigKey, configs.getString(connectConfigKey));
}
}
converterdConfigs.add(newKeyValue);
- connectTopicNames.add(configs.getString(RuntimeConfigDefine.CONNECT_TOPICNAME));
+ connectTopicNames.add(configs.getString(RuntimerConfigDefine.CONNECT_TOPICNAME));
}
putTaskConfigs(connectorName, converterdConfigs);
}
@@ -175,8 +174,8 @@
if(Objects.isNull(config)){
return;
}
- config.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, System.currentTimeMillis());
- config.put(RuntimeConfigDefine.CONFIG_DELETED, 1);
+ config.put(RuntimerConfigDefine.UPDATE_TIMESTAMP, System.currentTimeMillis());
+ config.put(RuntimerConfigDefine.CONFIG_DELETED, 1);
List<TargetKeyValue> taskConfigList = taskKeyValueStore.get(connectorName);
taskConfigList.add(config);
connectorKeyValueStore.put(connectorName, config);
