pandaapo commented on code in PR #4544:
URL: https://github.com/apache/eventmesh/pull/4544#discussion_r1388068099
##########
eventmesh-runtime/build.gradle:
##########
@@ -72,6 +72,9 @@ dependencies {
implementation project(":eventmesh-webhook:eventmesh-webhook-api")
implementation project(":eventmesh-webhook:eventmesh-webhook-receive")
+ implementation project(":eventmesh-retry:eventmesh-retry-api")
+ implementation project(":eventmesh-retry:eventmesh-retry-rocketmq")
Review Comment:
You can remove this dependency now.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/AbstractPushRequest.java:
##########
@@ -158,7 +158,7 @@ protected void removeWaitingMap(WebhookPushRequest request)
{
}
@Override
- public void run(Timeout timeout) throws Exception {
+ public void doRun(Timeout timeout) throws Exception {
tryPushRequest();
}
Review Comment:
Could `Timeout` in the `doRun (timeout)`<-- `run (timeout)` reserved for
`reput()` be removed ?
那为`rePut()`预留的`doRun(Timeout)` <-- `run(Timeout)`中的Timeout是不是可以删掉了？
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/RetryContext.java:
##########
@@ -17,43 +17,108 @@
package org.apache.eventmesh.runtime.core.protocol;
-import org.apache.eventmesh.runtime.core.timer.Timeout;
-import org.apache.eventmesh.runtime.core.timer.Timer;
-import org.apache.eventmesh.runtime.core.timer.TimerTask;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+import org.apache.eventmesh.retry.api.strategy.StorageRetryStrategy;
+import org.apache.eventmesh.retry.api.timer.Timeout;
+import org.apache.eventmesh.retry.api.timer.TimerTask;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
+import
org.apache.eventmesh.runtime.core.protocol.consumer.HandleMessageContext;
+import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
+import org.apache.eventmesh.runtime.core.protocol.producer.ProducerManager;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
-import java.util.concurrent.TimeUnit;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
import io.cloudevents.CloudEvent;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
public abstract class RetryContext implements TimerTask {
+ private static final Set<String> STORAGE_RETRY_PROCESSED_EVENT_LIST =
Collections.synchronizedSet(new HashSet<>());
+
public CloudEvent event;
public String seq;
public int retryTimes;
+ public CommonConfiguration commonConfiguration;
+
public long executeTime = System.currentTimeMillis();
- public long getExecuteTime() {
- return executeTime;
+ public void setEvent(CloudEvent event) {
+ this.event = event;
}
- protected void rePut(Timeout timeout, long tick, TimeUnit timeUnit) {
- if (timeout == null) {
- return;
- }
+ @Override
+ public void setExecuteTimeHook(long executeTime) {
+ this.executeTime = executeTime;
+ }
- Timer timer = timeout.timer();
- if (timer.isStop() || timeout.isCancelled()) {
+ @Override
+ public final void run(Timeout timeout) throws Exception {
+ boolean eventMeshServerRetryStorageEnabled =
+
Optional.ofNullable(commonConfiguration).map(CommonConfiguration::isEventMeshServerRetryStorageEnabled)
+ .orElse(false);
+ String eventMeshStoragePluginType =
commonConfiguration.getEventMeshStoragePluginType();
+ if (Constants.STANDALONE.equals(eventMeshStoragePluginType)) {
+ log.warn("Storage retry disabled, because of eventmesh storage is
standalone.");
Review Comment:
Perhaps "Because eventmesh storage is standalone, 'retry. storageEnable'
will be ignored" would be better, because it is not determined whether `retry.
storageEnable` is true here.
也许“Because eventmesh storage is standalone, 'retry.storageEnable' will be
ignored.”这样表述更好一些，因为这里并没判断`retry.storageEnable`是否为true。
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/RetryContext.java:
##########
@@ -17,43 +17,108 @@
package org.apache.eventmesh.runtime.core.protocol;
-import org.apache.eventmesh.runtime.core.timer.Timeout;
-import org.apache.eventmesh.runtime.core.timer.Timer;
-import org.apache.eventmesh.runtime.core.timer.TimerTask;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+import org.apache.eventmesh.retry.api.strategy.StorageRetryStrategy;
+import org.apache.eventmesh.retry.api.timer.Timeout;
+import org.apache.eventmesh.retry.api.timer.TimerTask;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
+import
org.apache.eventmesh.runtime.core.protocol.consumer.HandleMessageContext;
+import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
+import org.apache.eventmesh.runtime.core.protocol.producer.ProducerManager;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
-import java.util.concurrent.TimeUnit;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
import io.cloudevents.CloudEvent;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
public abstract class RetryContext implements TimerTask {
+ private static final Set<String> STORAGE_RETRY_PROCESSED_EVENT_LIST =
Collections.synchronizedSet(new HashSet<>());
+
public CloudEvent event;
public String seq;
public int retryTimes;
+ public CommonConfiguration commonConfiguration;
+
public long executeTime = System.currentTimeMillis();
- public long getExecuteTime() {
- return executeTime;
+ public void setEvent(CloudEvent event) {
+ this.event = event;
}
- protected void rePut(Timeout timeout, long tick, TimeUnit timeUnit) {
- if (timeout == null) {
- return;
- }
+ @Override
+ public void setExecuteTimeHook(long executeTime) {
+ this.executeTime = executeTime;
+ }
- Timer timer = timeout.timer();
- if (timer.isStop() || timeout.isCancelled()) {
+ @Override
+ public final void run(Timeout timeout) throws Exception {
+ boolean eventMeshServerRetryStorageEnabled =
+
Optional.ofNullable(commonConfiguration).map(CommonConfiguration::isEventMeshServerRetryStorageEnabled)
+ .orElse(false);
+ String eventMeshStoragePluginType =
commonConfiguration.getEventMeshStoragePluginType();
+ if (Constants.STANDALONE.equals(eventMeshStoragePluginType)) {
+ log.warn("Storage retry disabled, because of eventmesh storage is
standalone.");
+ doRun(timeout);
return;
}
+ if (eventMeshServerRetryStorageEnabled) {
+ Optional<StorageRetryStrategy> storageRetryStrategy =
Optional.ofNullable(
+ EventMeshExtensionFactory.getExtension(RetryStrategy.class,
+
commonConfiguration.getEventMeshStoragePluginType()).getStorageRetryStrategy());
Review Comment:
Can the design of this SPI omit the `RetryStrategy` interface and directly
use `StorageRetryStrategy` as the SPI interface?
这个SPI的设计，是不是可以省去`RetryStrategy`接口，直接用`StorageRetryStrategy`作为SPI接口？
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/AbstractPushRequest.java:
##########
@@ -158,7 +158,7 @@ protected void removeWaitingMap(WebhookPushRequest request)
{
}
@Override
- public void run(Timeout timeout) throws Exception {
+ public void doRun(Timeout timeout) throws Exception {
tryPushRequest();
}
Review Comment:
Could `Timeout` in the `doRun (timeout)`<-- `run (timeout)` reserved for
`reput()` be removed ?
那为`rePut()`预留的`doRun(Timeout)` <-- `run(Timeout)`中的Timeout是不是可以删掉了？
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/RetryContext.java:
##########
@@ -17,43 +17,108 @@
package org.apache.eventmesh.runtime.core.protocol;
-import org.apache.eventmesh.runtime.core.timer.Timeout;
-import org.apache.eventmesh.runtime.core.timer.Timer;
-import org.apache.eventmesh.runtime.core.timer.TimerTask;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+import org.apache.eventmesh.retry.api.strategy.StorageRetryStrategy;
+import org.apache.eventmesh.retry.api.timer.Timeout;
+import org.apache.eventmesh.retry.api.timer.TimerTask;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
+import
org.apache.eventmesh.runtime.core.protocol.consumer.HandleMessageContext;
+import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
+import org.apache.eventmesh.runtime.core.protocol.producer.ProducerManager;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
-import java.util.concurrent.TimeUnit;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
import io.cloudevents.CloudEvent;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
public abstract class RetryContext implements TimerTask {
+ private static final Set<String> STORAGE_RETRY_PROCESSED_EVENT_LIST =
Collections.synchronizedSet(new HashSet<>());
+
public CloudEvent event;
public String seq;
public int retryTimes;
+ public CommonConfiguration commonConfiguration;
+
public long executeTime = System.currentTimeMillis();
- public long getExecuteTime() {
- return executeTime;
+ public void setEvent(CloudEvent event) {
+ this.event = event;
}
- protected void rePut(Timeout timeout, long tick, TimeUnit timeUnit) {
- if (timeout == null) {
- return;
- }
+ @Override
+ public void setExecuteTimeHook(long executeTime) {
+ this.executeTime = executeTime;
+ }
- Timer timer = timeout.timer();
- if (timer.isStop() || timeout.isCancelled()) {
+ @Override
+ public final void run(Timeout timeout) throws Exception {
+ boolean eventMeshServerRetryStorageEnabled =
+
Optional.ofNullable(commonConfiguration).map(CommonConfiguration::isEventMeshServerRetryStorageEnabled)
+ .orElse(false);
+ String eventMeshStoragePluginType =
commonConfiguration.getEventMeshStoragePluginType();
+ if (Constants.STANDALONE.equals(eventMeshStoragePluginType)) {
+ log.warn("Storage retry disabled, because of eventmesh storage is
standalone.");
Review Comment:
Perhaps "Because eventmesh storage is standalone, 'retry. storageEnable'
will be ignored" would be better, because it is not determined whether `retry.
storageEnable` is true here.
也许“Because eventmesh storage is standalone, 'retry.storageEnable' will be
ignored.”这样表述更好一些，因为这里并没判断`retry.storageEnable`是否为true。
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/RetryContext.java:
##########
@@ -17,43 +17,108 @@
package org.apache.eventmesh.runtime.core.protocol;
-import org.apache.eventmesh.runtime.core.timer.Timeout;
-import org.apache.eventmesh.runtime.core.timer.Timer;
-import org.apache.eventmesh.runtime.core.timer.TimerTask;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+import org.apache.eventmesh.retry.api.strategy.StorageRetryStrategy;
+import org.apache.eventmesh.retry.api.timer.Timeout;
+import org.apache.eventmesh.retry.api.timer.TimerTask;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
+import
org.apache.eventmesh.runtime.core.protocol.consumer.HandleMessageContext;
+import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
+import org.apache.eventmesh.runtime.core.protocol.producer.ProducerManager;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
-import java.util.concurrent.TimeUnit;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
import io.cloudevents.CloudEvent;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
public abstract class RetryContext implements TimerTask {
+ private static final Set<String> STORAGE_RETRY_PROCESSED_EVENT_LIST =
Collections.synchronizedSet(new HashSet<>());
+
public CloudEvent event;
public String seq;
public int retryTimes;
+ public CommonConfiguration commonConfiguration;
+
public long executeTime = System.currentTimeMillis();
- public long getExecuteTime() {
- return executeTime;
+ public void setEvent(CloudEvent event) {
+ this.event = event;
}
- protected void rePut(Timeout timeout, long tick, TimeUnit timeUnit) {
- if (timeout == null) {
- return;
- }
+ @Override
+ public void setExecuteTimeHook(long executeTime) {
+ this.executeTime = executeTime;
+ }
- Timer timer = timeout.timer();
- if (timer.isStop() || timeout.isCancelled()) {
+ @Override
+ public final void run(Timeout timeout) throws Exception {
+ boolean eventMeshServerRetryStorageEnabled =
+
Optional.ofNullable(commonConfiguration).map(CommonConfiguration::isEventMeshServerRetryStorageEnabled)
+ .orElse(false);
+ String eventMeshStoragePluginType =
commonConfiguration.getEventMeshStoragePluginType();
+ if (Constants.STANDALONE.equals(eventMeshStoragePluginType)) {
+ log.warn("Storage retry disabled, because of eventmesh storage is
standalone.");
+ doRun(timeout);
return;
}
+ if (eventMeshServerRetryStorageEnabled) {
+ Optional<StorageRetryStrategy> storageRetryStrategy =
Optional.ofNullable(
+ EventMeshExtensionFactory.getExtension(RetryStrategy.class,
+
commonConfiguration.getEventMeshStoragePluginType()).getStorageRetryStrategy());
+ if (!storageRetryStrategy.isPresent()) {
+ log.warn("Storage retry SPI not found, retry in memory.");
+ doRun(timeout);
+ return;
+ }
+ if (!STORAGE_RETRY_PROCESSED_EVENT_LIST.contains(event.getId())) {
+ String consumerGroupName =
getHandleMessageContext().getConsumerGroup();
+ EventMeshProducer producer =
getProducerManager().getEventMeshProducer(consumerGroupName);
+ RetryConfiguration retryConfiguration =
RetryConfiguration.builder()
+ .event(event)
+ .retryStorageEnabled(eventMeshServerRetryStorageEnabled)
+ .consumerGroupName(consumerGroupName)
+
.producer(producer.getMqProducerWrapper().getMeshMQProducer())
+ .topic(getHandleMessageContext().getTopic())
+ .build();
+ storageRetryStrategy.get().retry(retryConfiguration);
+ STORAGE_RETRY_PROCESSED_EVENT_LIST.add(event.getId());
+ } else {
+ STORAGE_RETRY_PROCESSED_EVENT_LIST.remove(event.getId());
+ getHandleMessageContext().finish();
+ }
+ } else {
+ log.warn("Storage retry disabled, retry in memory.");
Review Comment:
`log.info` is better. Because this logical branch comes from
`retry.storageEnable=false`.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/RetryContext.java:
##########
@@ -17,43 +17,108 @@
package org.apache.eventmesh.runtime.core.protocol;
-import org.apache.eventmesh.runtime.core.timer.Timeout;
-import org.apache.eventmesh.runtime.core.timer.Timer;
-import org.apache.eventmesh.runtime.core.timer.TimerTask;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+import org.apache.eventmesh.retry.api.strategy.StorageRetryStrategy;
+import org.apache.eventmesh.retry.api.timer.Timeout;
+import org.apache.eventmesh.retry.api.timer.TimerTask;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
+import
org.apache.eventmesh.runtime.core.protocol.consumer.HandleMessageContext;
+import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
+import org.apache.eventmesh.runtime.core.protocol.producer.ProducerManager;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
-import java.util.concurrent.TimeUnit;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
import io.cloudevents.CloudEvent;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
public abstract class RetryContext implements TimerTask {
+ private static final Set<String> STORAGE_RETRY_PROCESSED_EVENT_LIST =
Collections.synchronizedSet(new HashSet<>());
+
public CloudEvent event;
public String seq;
public int retryTimes;
+ public CommonConfiguration commonConfiguration;
+
public long executeTime = System.currentTimeMillis();
- public long getExecuteTime() {
- return executeTime;
+ public void setEvent(CloudEvent event) {
+ this.event = event;
}
- protected void rePut(Timeout timeout, long tick, TimeUnit timeUnit) {
- if (timeout == null) {
- return;
- }
+ @Override
+ public void setExecuteTimeHook(long executeTime) {
+ this.executeTime = executeTime;
+ }
- Timer timer = timeout.timer();
- if (timer.isStop() || timeout.isCancelled()) {
+ @Override
+ public final void run(Timeout timeout) throws Exception {
+ boolean eventMeshServerRetryStorageEnabled =
+
Optional.ofNullable(commonConfiguration).map(CommonConfiguration::isEventMeshServerRetryStorageEnabled)
+ .orElse(false);
+ String eventMeshStoragePluginType =
commonConfiguration.getEventMeshStoragePluginType();
+ if (Constants.STANDALONE.equals(eventMeshStoragePluginType)) {
+ log.warn("Storage retry disabled, because of eventmesh storage is
standalone.");
+ doRun(timeout);
return;
}
+ if (eventMeshServerRetryStorageEnabled) {
+ Optional<StorageRetryStrategy> storageRetryStrategy =
Optional.ofNullable(
+ EventMeshExtensionFactory.getExtension(RetryStrategy.class,
+
commonConfiguration.getEventMeshStoragePluginType()).getStorageRetryStrategy());
Review Comment:
Can the design of this SPI omit the `RetryStrategy` interface and directly
use `StorageRetryStrategy` as the SPI interface?
这个SPI的设计，是不是可以省去`RetryStrategy`接口，直接用`StorageRetryStrategy`作为SPI接口？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
