[ISSUE#4178] When there are a large number of errors in the HTTP targ…

…et, RocketMQ consumption is slow (#4544)

* Provide MQ-storage-based retry strategy in http protocol.

* Remove unused method.

* fix

* fix

* Add retry, rocketmq-retry module.

* Do some optimization.

* Use interface instead of HttpHandleMsgContext.

* optimize

* fix

* Remove unused methods.

* fix log

* Remove retry-rocketmq dependency.

* rollback dependency changes.

* Use plugin to load spi.

* Remove unused methods.

* fix: remove storage dependency

* fix: log

* fix: change 'standalone' to 'default'

* fix: codereview

eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java

@@ -204,4 +204,6 @@ public class Constants {
 
     public static final String OS_WIN_PREFIX = "win";
 
+    public static final String DEFAULT = "default";
+
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java

@@ -19,6 +19,7 @@
 
 import static org.apache.eventmesh.common.Constants.HTTP;
 
+import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.common.utils.IPUtils;
 
 import org.apache.commons.collections4.CollectionUtils;
@@ -111,6 +112,9 @@ public class CommonConfiguration {
     @ConfigFiled(reload = true)
     private String meshGroup;
 
+    @ConfigFiled(field = "server.retry.plugin.type")
+    private String eventMeshRetryPluginType = Constants.DEFAULT;
+
     public void reload() {
         this.eventMeshWebhookOrigin = "eventmesh." + eventMeshIDC;
 

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/ProtocolKey.java

@@ -32,6 +32,7 @@ public class ProtocolKey {
     public static final String PROTOCOL_VERSION = "protocolversion";
 
     public static final String PROTOCOL_DESC = "protocoldesc";
+    public static final String TOPIC = "topic";
 
     public static final String CONTENT_TYPE = "contenttype";
 

eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ReflectUtils.java

@@ -20,6 +20,9 @@
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
 
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
 public class ReflectUtils {
 
     /**

eventmesh-retry/eventmesh-retry-api/build.gradle

@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+dependencies {
+    implementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
+    implementation project(":eventmesh-common")
+    implementation project(":eventmesh-spi")
+
+    implementation 'io.cloudevents:cloudevents-core'
+    implementation 'io.cloudevents:cloudevents-json-jackson'
+
+    compileOnly 'org.projectlombok:lombok'
+    annotationProcessor 'org.projectlombok:lombok'
+
+    testImplementation 'org.junit.jupiter:junit-jupiter'
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/AbstractRetryer.java → eventmesh-retry/eventmesh-retry-api/src/main/java/org/apache/eventmesh/retry/api/AbstractRetryer.java

@@ -15,13 +15,12 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol;
+package org.apache.eventmesh.retry.api;
 
 import org.apache.eventmesh.common.EventMeshThreadFactory;
-import org.apache.eventmesh.runtime.core.retry.Retryer;
-import org.apache.eventmesh.runtime.core.timer.HashedWheelTimer;
-import org.apache.eventmesh.runtime.core.timer.Timer;
-import org.apache.eventmesh.runtime.core.timer.TimerTask;
+import org.apache.eventmesh.retry.api.timer.HashedWheelTimer;
+import org.apache.eventmesh.retry.api.timer.Timer;
+import org.apache.eventmesh.retry.api.timer.TimerTask;
 
 import java.text.SimpleDateFormat;
 import java.util.Date;

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/retry/Retryer.java → eventmesh-retry/eventmesh-retry-api/src/main/java/org/apache/eventmesh/retry/api/Retryer.java

@@ -15,9 +15,9 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.retry;
+package org.apache.eventmesh.retry.api;
 
-import org.apache.eventmesh.runtime.core.timer.TimerTask;
+import org.apache.eventmesh.retry.api.timer.TimerTask;
 
 import java.util.concurrent.TimeUnit;
 

eventmesh-retry/eventmesh-retry-api/src/main/java/org/apache/eventmesh/retry/api/conf/RetryConfiguration.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.eventmesh.retry.api.conf;
+
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.SubscriptionMode;
+
+import io.cloudevents.CloudEvent;
+
+import lombok.Builder;
+import lombok.Data;
+
+@Data
+@Builder
+public class RetryConfiguration {
+
+    private CloudEvent event;
+
+    private String consumerGroupName;
+
+    private Producer producer;
+
+    private String topic;
+
+    private SubscriptionMode subscriptionMode;
+}

eventmesh-retry/eventmesh-retry-api/src/main/java/org/apache/eventmesh/retry/api/strategy/RetryStrategy.java

@@ -0,0 +1,31 @@
+/*
+ * 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.eventmesh.retry.api.strategy;
+
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.spi.EventMeshExtensionType;
+import org.apache.eventmesh.spi.EventMeshSPI;
+
+/**
+ * Retry strategy.
+ */
+@EventMeshSPI(isSingleton = false, eventMeshExtensionType = EventMeshExtensionType.RETRY)
+public interface RetryStrategy {
+
+    void retry(RetryConfiguration configuration);
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/timer/HashedWheelTimer.java → eventmesh-retry/eventmesh-retry-api/src/main/java/org/apache/eventmesh/retry/api/timer/HashedWheelTimer.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.timer;
+package org.apache.eventmesh.retry.api.timer;
 
 import static org.apache.eventmesh.common.Constants.OS_NAME_KEY;
 import static org.apache.eventmesh.common.Constants.OS_WIN_PREFIX;
@@ -632,7 +632,7 @@ public void expire() {
             }
 
             try {
-                task.run(this);
+                task.run();
                 task.setExecuteTimeHook(System.currentTimeMillis());
             } catch (Throwable t) {
                 if (logger.isWarnEnabled()) {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/timer/Timeout.java → eventmesh-retry/eventmesh-retry-api/src/main/java/org/apache/eventmesh/retry/api/timer/Timeout.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.timer;
+package org.apache.eventmesh.retry.api.timer;
 
 /**
  * A handle associated with a {@link TimerTask} that is returned by a

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/timer/Timer.java → eventmesh-retry/eventmesh-retry-api/src/main/java/org/apache/eventmesh/retry/api/timer/Timer.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.timer;
+package org.apache.eventmesh.retry.api.timer;
 
 import java.util.Set;
 import java.util.concurrent.RejectedExecutionException;

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/timer/TimerTask.java → eventmesh-retry/eventmesh-retry-api/src/main/java/org/apache/eventmesh/retry/api/timer/TimerTask.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.timer;
+package org.apache.eventmesh.retry.api.timer;
 
 import java.util.concurrent.TimeUnit;
 
@@ -28,10 +28,8 @@ public interface TimerTask {
     /**
      * Executed after the delay specified with
      * {@link Timer#newTimeout(TimerTask, long, TimeUnit)}.
-     *
-     * @param timeout a handle which is associated with this task
      */
-    void run(Timeout timeout) throws Exception;
+    void run() throws Exception;
 
     /**
      * Hook method to set the execute time.

eventmesh-retry/eventmesh-retry-rocketmq/build.gradle

@@ -0,0 +1,47 @@
+/*
+ * 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.
+ */
+
+List rocketmq = [
+        "org.apache.rocketmq:rocketmq-client:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-broker:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-common:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-store:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-namesrv:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-tools:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-remoting:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-logging:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-srvutil:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-filter:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-acl:$rocketmq_version",
+        "org.apache.rocketmq:rocketmq-srvutil:$rocketmq_version",
+]
+
+dependencies {
+    implementation project(":eventmesh-storage-plugin:eventmesh-storage-api")
+    implementation project(":eventmesh-storage-plugin:eventmesh-storage-rocketmq")
+    implementation rocketmq
+    implementation project(":eventmesh-retry:eventmesh-retry-api")
+    implementation project(":eventmesh-common")
+
+    implementation 'io.cloudevents:cloudevents-core'
+    implementation 'io.cloudevents:cloudevents-json-jackson'
+
+    compileOnly 'org.projectlombok:lombok'
+    annotationProcessor 'org.projectlombok:lombok'
+
+    testImplementation 'org.junit.jupiter:junit-jupiter'
+}

eventmesh-retry/eventmesh-retry-rocketmq/gradle.properties

@@ -0,0 +1,19 @@
+# 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.
+#
+
+rocketmq_version=4.9.5
+pluginType=retry
+pluginName=rocketmq

eventmesh-retry/eventmesh-retry-rocketmq/src/main/java/org/apache/eventmesh/retry/rocketmq/RocketMQRetryStrategyImpl.java

@@ -0,0 +1,75 @@
+/*
+ * 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.eventmesh.retry.rocketmq;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.rocketmq.common.MixAll;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RocketMQRetryStrategyImpl implements RetryStrategy {
+
+    @Override
+    public void retry(RetryConfiguration configuration) {
+        sendMessageBack(configuration);
+    }
+
+    @SneakyThrows
+    private void sendMessageBack(final RetryConfiguration configuration) {
+        CloudEvent event = configuration.getEvent();
+        String topic = configuration.getTopic();
+        String consumerGroupName = configuration.getConsumerGroupName();
+        String retryTopicName = MixAll.getRetryTopic(consumerGroupName);
+
+        String bizSeqNo = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.BIZSEQNO.getKey())).toString();
+        String uniqueId = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.UNIQUEID.getKey())).toString();
+        CloudEvent retryEvent = CloudEventBuilder.from(event)
+            .withExtension(ProtocolKey.TOPIC, topic)
+            .withSubject(retryTopicName)
+            .build();
+        Producer producer = configuration.getProducer();
+        producer.publish(retryEvent, new SendCallback() {
+
+            @Override
+            public void onSuccess(SendResult sendResult) {
+                log.info("consumer:{} consume success,, bizSeqno:{}, uniqueId:{}",
+                    consumerGroupName, bizSeqNo, uniqueId);
+            }
+
+            @Override
+            public void onException(OnExceptionContext context) {
+                log.warn("consumer:{} consume fail, sendMessageBack, bizSeqno:{}, uniqueId:{}",
+                    consumerGroupName, bizSeqNo, uniqueId, context.getException());
+            }
+        });
+    }
+}

eventmesh-retry/eventmesh-retry-rocketmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.retry.api.strategy.RetryStrategy

@@ -0,0 +1,16 @@
+# 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.
+
+rocketmq=org.apache.eventmesh.retry.rocketmq.RocketMQRetryStrategyImpl