You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by sh...@apache.org on 2023/04/24 03:03:07 UTC
[rocketmq-eventbridge] branch runtimer updated (2f03d44 -> f12cc04)
This is an automated email from the ASF dual-hosted git repository.
shenlin pushed a change to branch runtimer
in repository https://gitbox.apache.org/repos/asf/rocketmq-eventbridge.git
from 2f03d44 Add simulated backpressure test code
new 667111e feat:support record commit.
new f12cc04 feat:support record commit.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../eventbridge/adapter/runtimer/Runtimer.java | 45 ++++++------
.../adapter/runtimer/boot/EventBusListener.java | 18 +++--
.../adapter/runtimer/boot/EventRuleTransfer.java | 82 +++++++++++++---------
.../adapter/runtimer/boot/EventTargetPusher.java | 24 ++++---
.../PushCallback.java => OffsetManager.java} | 18 +++--
.../runtimer/boot/listener/EventSubscriber.java | 11 +++
.../boot/listener/RocketMQEventSubscriber.java | 4 ++
.../adapter/runtimer/boot/pusher/PushRequest.java | 34 ---------
.../adapter/runtimer/common/ServiceThread.java | 1 +
.../runtimer/config/RuntimerConfiguration.java | 41 +++++++++++
.../{retry => error}/DeadLetterQueueService.java | 2 +-
.../runtimer/{retry => error}/ErrorHandler.java | 16 +++--
.../adapter/runtimer/retry/EventBusStorage.java | 25 -------
.../runtimer/retry/EventBusStorageOnRocketMQ.java | 34 ---------
.../service/TargetRunnerConfigOnFileObserver.java | 2 +
15 files changed, 184 insertions(+), 173 deletions(-)
rename adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/{pusher/PushCallback.java => OffsetManager.java} (68%)
delete mode 100644 adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/pusher/PushRequest.java
create mode 100644 adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimerConfiguration.java
rename adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/{retry => error}/DeadLetterQueueService.java (92%)
rename adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/{retry => error}/ErrorHandler.java (87%)
delete mode 100644 adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/EventBusStorage.java
delete mode 100644 adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/EventBusStorageOnRocketMQ.java
[rocketmq-eventbridge] 02/02: feat:support record commit.
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shenlin pushed a commit to branch runtimer
in repository https://gitbox.apache.org/repos/asf/rocketmq-eventbridge.git
commit f12cc0449ef525bbd6215951bc35756b8a513dd6
Author: 2011shenlin <20...@gmail.com>
AuthorDate: Sun Apr 23 19:47:27 2023 +0800
feat:support record commit.
---
.../eventbridge/adapter/runtimer/Runtimer.java | 15 ++++++---
.../adapter/runtimer/boot/EventBusListener.java | 18 ++++++----
.../adapter/runtimer/boot/EventRuleTransfer.java | 14 +++++---
.../adapter/runtimer/boot/EventTargetPusher.java | 6 +++-
.../runtimer/boot/listener/EventSubscriber.java | 3 +-
.../adapter/runtimer/boot/pusher/PushCallback.java | 39 ----------------------
.../adapter/runtimer/boot/pusher/PushRequest.java | 34 -------------------
.../adapter/runtimer/error/ErrorHandler.java | 7 +++-
8 files changed, 44 insertions(+), 92 deletions(-)
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 495aeee..0cf382b 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
@@ -26,6 +26,7 @@ import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.OffsetManager;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.CirculatorContext;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.EventSubscriber;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.RuntimerState;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.error.ErrorHandler;
import org.apache.rocketmq.eventbridge.adapter.runtimer.service.TargetRunnerConfigObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,24 +45,28 @@ public class Runtimer {
private AtomicReference<RuntimerState> runtimerState;
+ @Autowired
private CirculatorContext circulatorContext;
-
@Autowired
private TargetRunnerConfigObserver runnerConfigObserver;
@Autowired
private OffsetManager offsetManager;
@Autowired
private EventSubscriber eventSubscriber;
+ @Autowired
+ private ErrorHandler errorHandler;
public Runtimer(
CirculatorContext circulatorContext,
TargetRunnerConfigObserver runnerConfigObserver,
OffsetManager offsetManager,
- EventSubscriber eventSubscriber) {
+ EventSubscriber eventSubscriber,
+ ErrorHandler errorHandler) {
this.circulatorContext = circulatorContext;
this.runnerConfigObserver = runnerConfigObserver;
this.offsetManager = offsetManager;
this.eventSubscriber = eventSubscriber;
+ this.errorHandler = errorHandler;
}
@PostConstruct
@@ -70,9 +75,9 @@ public class Runtimer {
circulatorContext.initListenerMetadata(runnerConfigObserver.getTargetRunnerConfig());
runnerConfigObserver.registerListener(circulatorContext);
runnerConfigObserver.registerListener(eventSubscriber);
- new EventBusListener(circulatorContext, eventSubscriber).start();
- new EventRuleTransfer(circulatorContext, offsetManager).start();
- new EventTargetPusher(circulatorContext, offsetManager).start();
+ new EventBusListener(circulatorContext, eventSubscriber, errorHandler).start();
+ new EventRuleTransfer(circulatorContext, offsetManager, errorHandler).start();
+ new EventTargetPusher(circulatorContext, offsetManager, errorHandler).start();
startRuntimer();
}
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 7a34aa3..d735e9a 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
@@ -18,15 +18,15 @@
package org.apache.rocketmq.eventbridge.adapter.runtimer.boot;
import io.openmessaging.connector.api.data.ConnectRecord;
+import java.util.List;
import org.apache.commons.collections.CollectionUtils;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.EventSubscriber;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.CirculatorContext;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.EventSubscriber;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.ServiceThread;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.error.ErrorHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.List;
-
/**
* listen the event and offer to queue
*
@@ -40,22 +40,26 @@ public class EventBusListener extends ServiceThread {
private final EventSubscriber eventSubscriber;
- public EventBusListener(CirculatorContext circulatorContext, EventSubscriber eventSubscriber) {
+ private final ErrorHandler errorHandler;
+
+ public EventBusListener(CirculatorContext circulatorContext, EventSubscriber eventSubscriber,
+ ErrorHandler errorHandler) {
this.circulatorContext = circulatorContext;
this.eventSubscriber = eventSubscriber;
+ this.errorHandler = errorHandler;
}
@Override
public void run() {
while (!stopped) {
- try{
+ try {
List<ConnectRecord> recordList = eventSubscriber.pull();
- if(CollectionUtils.isEmpty(recordList)){
+ if (CollectionUtils.isEmpty(recordList)) {
this.waitForRunning(1000);
continue;
}
circulatorContext.offerEventRecords(recordList);
- }catch (Exception exception) {
+ } catch (Exception exception) {
logger.error(getServiceName() + " - event bus pull record exception, stackTrace - ", exception);
}
}
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 06f1ce3..a094d07 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
@@ -33,10 +33,10 @@ import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.Circulator
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.transfer.TransformEngine;
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.error.ErrorHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
/**
* receive event and transfer the rule to pusher
@@ -49,10 +49,14 @@ public class EventRuleTransfer extends ServiceThread {
private final CirculatorContext circulatorContext;
@Autowired
private final OffsetManager offsetManager;
+ @Autowired
+ private final ErrorHandler errorHandler;
- public EventRuleTransfer(CirculatorContext circulatorContext, OffsetManager offsetManager) {
+ public EventRuleTransfer(CirculatorContext circulatorContext, OffsetManager offsetManager,
+ ErrorHandler errorHandler) {
this.circulatorContext = circulatorContext;
this.offsetManager = offsetManager;
+ this.errorHandler = errorHandler;
}
@Override
@@ -61,9 +65,10 @@ public class EventRuleTransfer extends ServiceThread {
}
@PostConstruct
- public void init(){
+ public void init() {
super.start();
}
+
@Override
public void run() {
while (!stopped) {
@@ -96,6 +101,7 @@ public class EventRuleTransfer extends ServiceThread {
CompletableFuture<Void> transformFuture = CompletableFuture.supplyAsync(() -> transfer.doTransforms(eventRecord))
.exceptionally((exception) -> {
logger.error("transfer do transform event record failed，stackTrace-", exception);
+ errorHandler.handle(eventRecord,exception);
return null;
})
.thenAccept(record -> {
@@ -105,7 +111,7 @@ public class EventRuleTransfer extends ServiceThread {
record.getExtensions().put(runnerNameKey, transfer.getConnectConfig(runnerNameKey));
record.getExtensions().put(taskClassKey, transfer.getConnectConfig(taskClassKey));
afterTransformConnect.add(record);
- }else{
+ } else {
offsetManager.commit(eventRecord);
}
});
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 e420ee2..b3216da 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
@@ -26,6 +26,7 @@ import java.util.concurrent.ExecutorService;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.CirculatorContext;
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.error.ErrorHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,10 +41,13 @@ public class EventTargetPusher extends ServiceThread {
private final CirculatorContext circulatorContext;
private final OffsetManager offsetManager;
+ private final ErrorHandler errorHandler;
- public EventTargetPusher(CirculatorContext circulatorContext, OffsetManager offsetManager) {
+ public EventTargetPusher(CirculatorContext circulatorContext, OffsetManager offsetManager,
+ ErrorHandler errorHandler) {
this.circulatorContext = circulatorContext;
this.offsetManager = offsetManager;
+ this.errorHandler = errorHandler;
}
@Override
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/EventSubscriber.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/EventSubscriber.java
index 39d08d7..0b33b9e 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/EventSubscriber.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/EventSubscriber.java
@@ -52,8 +52,9 @@ public abstract class EventSubscriber implements TargetRunnerListener {
* @param connectRecord
* @param delaySec
*/
- public void put(String eventBusName, ConnectRecord connectRecord, int delaySec){
+ public boolean put(String eventBusName, ConnectRecord connectRecord, int delaySec){
// convert the eventBusName to Topic ?
+ return true;
}
/**
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/pusher/PushCallback.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/pusher/PushCallback.java
deleted file mode 100644
index 8609cad..0000000
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/pusher/PushCallback.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.rocketmq.eventbridge.adapter.runtimer.boot.pusher;
-
-import com.google.common.collect.Lists;
-import io.openmessaging.connector.api.data.ConnectRecord;
-import java.util.List;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.EventSubscriber;
-import org.springframework.beans.factory.annotation.Autowired;
-
-public class PushCallback {
-
- @Autowired
- EventSubscriber eventSubscriber;
-
- public void completed(ConnectRecord connectRecord) {
- eventSubscriber.commit(Lists.newArrayList(connectRecord));
- }
-
- public void completed(List<ConnectRecord> connectRecords) {
- eventSubscriber.commit(connectRecords);
- }
-
-}
\ No newline at end of file
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/pusher/PushRequest.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/pusher/PushRequest.java
deleted file mode 100644
index 9223a5f..0000000
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/pusher/PushRequest.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.rocketmq.eventbridge.adapter.runtimer.boot.pusher;
-
-import io.openmessaging.connector.api.data.ConnectRecord;
-import java.util.List;
-import lombok.Data;
-
-@Data
-public class PushRequest {
-
- private String targetRunnerName;
-
- private String targetClass;
-
- private List<ConnectRecord> connectRecords;
-
- private PushCallback pushCallback;
-}
\ No newline at end of file
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/ErrorHandler.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/ErrorHandler.java
index 16bdbb9..9aac1d5 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/ErrorHandler.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/ErrorHandler.java
@@ -20,6 +20,7 @@ package org.apache.rocketmq.eventbridge.adapter.runtimer.error;
import com.google.common.base.Strings;
import io.openmessaging.connector.api.data.ConnectRecord;
import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.OffsetManager;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.EventSubscriber;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.TargetRunnerContext;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetRunnerConfig;
@@ -37,16 +38,21 @@ public class ErrorHandler {
@Autowired
EventSubscriber eventSubscriber;
+ @Autowired
+ OffsetManager offsetManager;
+
public void handle(ConnectRecord connectRecord, Throwable t) {
String eventRunnerName = connectRecord.getExtension(RUNNER_NAME);
TargetRunnerConfig targetRunnerConfig = TargetRunnerContext.getTargetRunnerConfig(eventRunnerName);
String eventBusName = targetRunnerConfig.getEventBusName();
PushRetryStrategyEnum pushRetryStrategyEnum = PushRetryStrategyEnum.parse(targetRunnerConfig.getRunOptions().getRetryStrategy());
+
int retryTimes = parseRetryTimes(connectRecord);
int delaySec = calcDelaySec(retryTimes, pushRetryStrategyEnum);
if (delaySec > 0) {
eventSubscriber.put(eventBusName, connectRecord, delaySec);
}
+ offsetManager.commit(connectRecord);
}
private int parseRetryTimes(ConnectRecord connectRecord) {
@@ -71,7 +77,6 @@ public class ErrorHandler {
* @return
*/
private int calcDelaySec(int retryTimes, PushRetryStrategyEnum pushRetryStrategyEnum) {
-
switch (pushRetryStrategyEnum) {
case BACKOFF_RETRY:
if (retryTimes >= pushRetryStrategyEnum.getRetryTimes()) {
[rocketmq-eventbridge] 01/02: feat:support record commit.
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shenlin pushed a commit to branch runtimer
in repository https://gitbox.apache.org/repos/asf/rocketmq-eventbridge.git
commit 667111e6b998a7c8007aeee3d48b1b0245cfaabe
Author: 2011shenlin <20...@gmail.com>
AuthorDate: Sun Apr 23 16:36:29 2023 +0800
feat:support record commit.
---
.../eventbridge/adapter/runtimer/Runtimer.java | 36 +++++-----
.../adapter/runtimer/boot/EventRuleTransfer.java | 76 +++++++++++++---------
.../adapter/runtimer/boot/EventTargetPusher.java | 20 +++---
.../OffsetManager.java} | 25 +++++--
.../runtimer/boot/listener/EventSubscriber.java | 10 +++
.../boot/listener/RocketMQEventSubscriber.java | 4 ++
.../adapter/runtimer/common/ServiceThread.java | 1 +
.../runtimer/config/RuntimerConfiguration.java | 41 ++++++++++++
.../{retry => error}/DeadLetterQueueService.java | 2 +-
.../runtimer/{retry => error}/ErrorHandler.java | 9 ++-
.../adapter/runtimer/retry/EventBusStorage.java | 25 -------
.../service/TargetRunnerConfigOnFileObserver.java | 2 +
12 files changed, 155 insertions(+), 96 deletions(-)
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 7eb922d..495aeee 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
@@ -22,17 +22,14 @@ import javax.annotation.PostConstruct;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.EventBusListener;
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.OffsetManager;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.CirculatorContext;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.EventSubscriber;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.RocketMQEventSubscriber;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.RuntimerState;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.common.enums.ConfigModeEnum;
import org.apache.rocketmq.eventbridge.adapter.runtimer.service.TargetRunnerConfigObserver;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.service.TargetRunnerConfigOnDBObserver;
-import org.apache.rocketmq.eventbridge.adapter.runtimer.service.TargetRunnerConfigOnFileObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Value;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
@@ -49,30 +46,33 @@ public class Runtimer {
private CirculatorContext circulatorContext;
+ @Autowired
private TargetRunnerConfigObserver runnerConfigObserver;
+ @Autowired
+ private OffsetManager offsetManager;
+ @Autowired
+ private EventSubscriber eventSubscriber;
- public Runtimer(CirculatorContext circulatorContext, @Value("${rumtimer.config.mode}") String configMode) {
+ public Runtimer(
+ CirculatorContext circulatorContext,
+ TargetRunnerConfigObserver runnerConfigObserver,
+ OffsetManager offsetManager,
+ EventSubscriber eventSubscriber) {
this.circulatorContext = circulatorContext;
- switch (ConfigModeEnum.parse(configMode)) {
- case DB:
- this.runnerConfigObserver = new TargetRunnerConfigOnDBObserver();
- break;
- default:
- this.runnerConfigObserver = new TargetRunnerConfigOnFileObserver();
- break;
- }
+ this.runnerConfigObserver = runnerConfigObserver;
+ this.offsetManager = offsetManager;
+ this.eventSubscriber = eventSubscriber;
}
@PostConstruct
public void initAndStart() {
- logger.info("init runtimer task config");
+ logger.info("Start init runtimer.");
circulatorContext.initListenerMetadata(runnerConfigObserver.getTargetRunnerConfig());
- EventSubscriber eventSubscriber = new RocketMQEventSubscriber(runnerConfigObserver);
runnerConfigObserver.registerListener(circulatorContext);
runnerConfigObserver.registerListener(eventSubscriber);
new EventBusListener(circulatorContext, eventSubscriber).start();
- new EventRuleTransfer(circulatorContext).start();
- new EventTargetPusher(circulatorContext).start();
+ new EventRuleTransfer(circulatorContext, offsetManager).start();
+ new EventTargetPusher(circulatorContext, offsetManager).start();
startRuntimer();
}
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 635c4bd..06f1ce3 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
@@ -20,6 +20,13 @@ package org.apache.rocketmq.eventbridge.adapter.runtimer.boot;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import io.openmessaging.connector.api.data.ConnectRecord;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import javax.annotation.PostConstruct;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.CirculatorContext;
@@ -28,13 +35,8 @@ import org.apache.rocketmq.eventbridge.adapter.runtimer.common.ServiceThread;
import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.concurrent.*;
-import java.util.stream.Collectors;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
/**
* receive event and transfer the rule to pusher
@@ -43,10 +45,14 @@ public class EventRuleTransfer extends ServiceThread {
private static final Logger logger = LoggerFactory.getLogger(EventRuleTransfer.class);
+ @Autowired
private final CirculatorContext circulatorContext;
+ @Autowired
+ private final OffsetManager offsetManager;
- public EventRuleTransfer(CirculatorContext circulatorContext) {
+ public EventRuleTransfer(CirculatorContext circulatorContext, OffsetManager offsetManager) {
this.circulatorContext = circulatorContext;
+ this.offsetManager = offsetManager;
}
@Override
@@ -54,6 +60,10 @@ public class EventRuleTransfer extends ServiceThread {
return this.getClass().getSimpleName();
}
+ @PostConstruct
+ public void init(){
+ super.start();
+ }
@Override
public void run() {
while (!stopped) {
@@ -64,7 +74,7 @@ public class EventRuleTransfer extends ServiceThread {
continue;
}
Map<String, TransformEngine<ConnectRecord>> latestTransformMap = circulatorContext.getTaskTransformMap();
- if(MapUtils.isEmpty(latestTransformMap)){
+ if (MapUtils.isEmpty(latestTransformMap)) {
logger.warn("latest transform engine is empty, continue by curTime - {}", System.currentTimeMillis());
this.waitForRunning(3000);
continue;
@@ -73,38 +83,40 @@ public class EventRuleTransfer extends ServiceThread {
String eventChannelName = RuntimerConfigDefine.CHANNEL_NAME;
String eventChannel = eventRecord.getExtension(eventChannelName);
Set<TransformEngine<ConnectRecord>> adaptTransformSet = latestTransformMap.values().stream()
- .filter(engine -> eventChannel.equals(engine.getConnectConfig(eventChannelName)))
- .collect(Collectors.toSet());
- if(CollectionUtils.isEmpty(adaptTransformSet)){
- logger.warn("adapt specific topic ref transform engine is empty, eventChannelName- {}", eventChannel);
- this.waitForRunning(3000);
- continue;
+ .filter(engine -> eventChannel.equals(engine.getConnectConfig(eventChannelName)))
+ .collect(Collectors.toSet());
+ if (CollectionUtils.isEmpty(adaptTransformSet)) {
+ logger.warn("adapt specific topic ref transform engine is empty, eventChannelName- {}", eventChannel);
+ this.waitForRunning(3000);
+ continue;
}
List<ConnectRecord> afterTransformConnect = Lists.newArrayList();
List<CompletableFuture<Void>> completableFutures = Lists.newArrayList();
- adaptTransformSet.forEach(transfer->{
- CompletableFuture<Void> transformFuture = CompletableFuture.supplyAsync(()-> transfer.doTransforms(eventRecord))
- .exceptionally((exception) -> {
- logger.error("transfer do transform event record failed，stackTrace-", exception);
- return null;
- })
- .thenAccept(record-> {
- if(Objects.nonNull(record)){
- String runnerNameKey = RuntimerConfigDefine.RUNNER_NAME;
- String taskClassKey = RuntimerConfigDefine.TASK_CLASS;
- record.getExtensions().put(runnerNameKey, transfer.getConnectConfig(runnerNameKey));
- record.getExtensions().put(taskClassKey, transfer.getConnectConfig(taskClassKey));
- afterTransformConnect.add(record);
- }
- });
+ adaptTransformSet.forEach(transfer -> {
+ CompletableFuture<Void> transformFuture = CompletableFuture.supplyAsync(() -> transfer.doTransforms(eventRecord))
+ .exceptionally((exception) -> {
+ logger.error("transfer do transform event record failed，stackTrace-", exception);
+ return null;
+ })
+ .thenAccept(record -> {
+ if (Objects.nonNull(record)) {
+ String runnerNameKey = RuntimerConfigDefine.RUNNER_NAME;
+ String taskClassKey = RuntimerConfigDefine.TASK_CLASS;
+ record.getExtensions().put(runnerNameKey, transfer.getConnectConfig(runnerNameKey));
+ record.getExtensions().put(taskClassKey, transfer.getConnectConfig(taskClassKey));
+ afterTransformConnect.add(record);
+ }else{
+ offsetManager.commit(eventRecord);
+ }
+ });
completableFutures.add(transformFuture);
});
- try{
+ try {
CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[adaptTransformSet.size()])).get();
circulatorContext.offerTargetTaskQueue(afterTransformConnect);
logger.info("offer target task queues succeed, transforms - {}", JSON.toJSONString(afterTransformConnect));
- }catch (Exception exception){
+ } catch (Exception exception) {
logger.error("transfer event record failed, stackTrace-", exception);
}
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 43e6b29..e420ee2 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
@@ -21,29 +21,29 @@ import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import io.openmessaging.connector.api.component.task.sink.SinkTask;
import io.openmessaging.connector.api.data.ConnectRecord;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.CirculatorContext;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.ServiceThread;
import org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.ExecutorService;
-
/**
* event target push to sink task
*
* @author artisan
*/
-public class EventTargetPusher extends ServiceThread{
+public class EventTargetPusher extends ServiceThread {
private static final Logger logger = LoggerFactory.getLogger(EventTargetPusher.class);
private final CirculatorContext circulatorContext;
+ private final OffsetManager offsetManager;
- public EventTargetPusher(CirculatorContext circulatorContext) {
+ public EventTargetPusher(CirculatorContext circulatorContext, OffsetManager offsetManager) {
this.circulatorContext = circulatorContext;
+ this.offsetManager = offsetManager;
}
@Override
@@ -55,7 +55,7 @@ public class EventTargetPusher extends ServiceThread{
this.waitForRunning(1000);
continue;
}
- if(logger.isDebugEnabled()){
+ if (logger.isDebugEnabled()) {
logger.debug("start push content by pusher - {}", JSON.toJSONString(targetRecord));
}
@@ -63,9 +63,10 @@ public class EventTargetPusher extends ServiceThread{
executorService.execute(() -> {
try {
String runnerName = targetRecord.getExtensions().getString(RuntimerConfigDefine.RUNNER_NAME);
- SinkTask sinkTask = circulatorContext.getPusherTaskMap().get(runnerName);;
+ SinkTask sinkTask = circulatorContext.getPusherTaskMap().get(runnerName);
sinkTask.put(Lists.newArrayList(targetRecord));
- }catch (Exception exception){
+ offsetManager.commit(targetRecord);
+ } catch (Exception exception) {
logger.error(getServiceName() + " push target exception, record - " + targetRecord + " , stackTrace-", exception);
}
});
@@ -77,5 +78,4 @@ public class EventTargetPusher extends ServiceThread{
return EventTargetPusher.class.getSimpleName();
}
-
}
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/EventBusStorageOnRocketMQ.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/OffsetManager.java
similarity index 54%
rename from adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/EventBusStorageOnRocketMQ.java
rename to adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/OffsetManager.java
index 94d8d4a..0613efb 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/EventBusStorageOnRocketMQ.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/OffsetManager.java
@@ -15,20 +15,31 @@
* limitations under the License.
*/
-package org.apache.rocketmq.eventbridge.adapter.runtimer.retry;
+package org.apache.rocketmq.eventbridge.adapter.runtimer.boot;
+import com.google.common.collect.Lists;
import io.openmessaging.connector.api.data.ConnectRecord;
import java.util.List;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.EventSubscriber;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
-public class EventBusStorageOnRocketMQ implements EventBusStorage {
+@Component
+public class OffsetManager {
- @Override
- public void put(String eventBusName, ConnectRecord connectRecord, int delaySec) {
+ @Autowired
+ EventSubscriber eventSubscriber;
+ public OffsetManager(EventSubscriber eventSubscriber) {
+ this.eventSubscriber = eventSubscriber;
}
- public List<String> parseEventBusName(String eventBusName) {
- //TODO
- return null;
+ public void commit(final List<ConnectRecord> connectRecordList) {
+ this.eventSubscriber.commit(connectRecordList);
}
+
+ public void commit(final ConnectRecord connectRecord) {
+ this.eventSubscriber.commit(Lists.newArrayList(connectRecord));
+ }
+
}
\ No newline at end of file
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/EventSubscriber.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/EventSubscriber.java
index f24ce08..39d08d7 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/EventSubscriber.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/EventSubscriber.java
@@ -46,6 +46,16 @@ public abstract class EventSubscriber implements TargetRunnerListener {
*/
public abstract void commit(List<ConnectRecord> connectRecordList);
+ /**
+ * Put the connect record to the eventbus.
+ * @param eventBusName
+ * @param connectRecord
+ * @param delaySec
+ */
+ public void put(String eventBusName, ConnectRecord connectRecord, int delaySec){
+ // convert the eventBusName to Topic ?
+ }
+
/**
* Call when add new target runner to runtimer.
*
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/RocketMQEventSubscriber.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/RocketMQEventSubscriber.java
index aa263c7..b4c7527 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/RocketMQEventSubscriber.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/boot/listener/RocketMQEventSubscriber.java
@@ -49,6 +49,7 @@ import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.proxy.SocksProxyConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import java.nio.charset.StandardCharsets;
@@ -60,16 +61,19 @@ import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import org.springframework.stereotype.Component;
/**
* RocketMQ implement event subscriber
*/
+@Component
public class RocketMQEventSubscriber extends EventSubscriber {
private static final Logger logger = LoggerFactory.getLogger(RocketMQEventSubscriber.class);
private LitePullConsumer pullConsumer;
+ @Autowired
private final TargetRunnerConfigObserver runnerConfigObserver;
private Integer pullTimeOut;
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/common/ServiceThread.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/common/ServiceThread.java
index 33f4d9d..566d8e2 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/common/ServiceThread.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/common/ServiceThread.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.eventbridge.adapter.runtimer.common;
+import javax.annotation.PostConstruct;
import org.apache.rocketmq.common.CountDownLatch2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimerConfiguration.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimerConfiguration.java
new file mode 100644
index 0000000..7160cca
--- /dev/null
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/config/RuntimerConfiguration.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.eventbridge.adapter.runtimer.config;
+
+import org.apache.rocketmq.eventbridge.adapter.runtimer.common.enums.ConfigModeEnum;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.service.TargetRunnerConfigObserver;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.service.TargetRunnerConfigOnDBObserver;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.service.TargetRunnerConfigOnFileObserver;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RuntimerConfiguration {
+
+ @Bean(name = "runnerConfigObserver")
+ public TargetRunnerConfigObserver targetRunnerConfigObserver(@Value("${rumtimer.config.mode}") String configMode) {
+ switch (ConfigModeEnum.parse(configMode)) {
+ case DB:
+ return new TargetRunnerConfigOnDBObserver();
+ default:
+ return new TargetRunnerConfigOnFileObserver();
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/DeadLetterQueueService.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/DeadLetterQueueService.java
similarity index 92%
rename from adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/DeadLetterQueueService.java
rename to adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/DeadLetterQueueService.java
index ce6a642..fd108c9 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/DeadLetterQueueService.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/DeadLetterQueueService.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.rocketmq.eventbridge.adapter.runtimer.retry;
+package org.apache.rocketmq.eventbridge.adapter.runtimer.error;
public class DeadLetterQueueService {
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/ErrorHandler.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/ErrorHandler.java
similarity index 91%
rename from adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/ErrorHandler.java
rename to adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/ErrorHandler.java
index 412b538..16bdbb9 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/ErrorHandler.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/error/ErrorHandler.java
@@ -15,24 +15,27 @@
* limitations under the License.
*/
-package org.apache.rocketmq.eventbridge.adapter.runtimer.retry;
+package org.apache.rocketmq.eventbridge.adapter.runtimer.error;
import com.google.common.base.Strings;
import io.openmessaging.connector.api.data.ConnectRecord;
import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.EventSubscriber;
import org.apache.rocketmq.eventbridge.adapter.runtimer.boot.listener.TargetRunnerContext;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetRunnerConfig;
import org.apache.rocketmq.eventbridge.enums.PushRetryStrategyEnum;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
import static org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine.CONNECT_RECORDS_KEY;
import static org.apache.rocketmq.eventbridge.adapter.runtimer.config.RuntimerConfigDefine.RUNNER_NAME;
@Slf4j
+@Component
public class ErrorHandler {
@Autowired
- EventBusStorage eventBusStorage;
+ EventSubscriber eventSubscriber;
public void handle(ConnectRecord connectRecord, Throwable t) {
String eventRunnerName = connectRecord.getExtension(RUNNER_NAME);
@@ -42,7 +45,7 @@ public class ErrorHandler {
int retryTimes = parseRetryTimes(connectRecord);
int delaySec = calcDelaySec(retryTimes, pushRetryStrategyEnum);
if (delaySec > 0) {
- eventBusStorage.put(eventBusName, connectRecord, delaySec);
+ eventSubscriber.put(eventBusName, connectRecord, delaySec);
}
}
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/EventBusStorage.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/EventBusStorage.java
deleted file mode 100644
index cbec332..0000000
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/retry/EventBusStorage.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.rocketmq.eventbridge.adapter.runtimer.retry;
-
-import io.openmessaging.connector.api.data.ConnectRecord;
-
-public interface EventBusStorage {
-
- void put(String eventBusName, ConnectRecord connectRecord, int delaySec);
-}
\ No newline at end of file
diff --git a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/service/TargetRunnerConfigOnFileObserver.java b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/service/TargetRunnerConfigOnFileObserver.java
index 63f1e9d..791efc1 100644
--- a/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/service/TargetRunnerConfigOnFileObserver.java
+++ b/adapter/runtimer/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtimer/service/TargetRunnerConfigOnFileObserver.java
@@ -41,9 +41,11 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.eventbridge.adapter.runtimer.common.entity.TargetRunnerConfig;
import org.apache.rocketmq.eventbridge.exception.EventBridgeException;
+import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
@Slf4j
+@Component
public class TargetRunnerConfigOnFileObserver extends AbstractTargetRunnerConfigObserver {
private String pathName;