Add maxReconsumeTimes support .

spring-cloud-alibaba-examples/pom.xml

@@ -40,6 +40,7 @@
         <module>rocketmq-example/rocketmq-sql-consume-example</module>
         <module>rocketmq-example/rocketmq-example-common</module>
         <module>rocketmq-example/rocketmq-tx-example</module>
+        <module>rocketmq-example/rocketmq-retrieable-consume-example</module>
 
         <module>spring-cloud-bus-rocketmq-example</module>
         <module>spring-cloud-alibaba-sidecar-examples/spring-cloud-alibaba-sidecar-nacos-example</module>

spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md

@@ -811,6 +811,92 @@ public class RocketMQTxApplication {
 }
 ```
 
+## 最大重试消费示例
+
+- 重试消费消息：根据配置的重新消费的次数，服务端会根据客户端消费是否成功，进行重新推送消息。
+
+### 创建Topic
+
+```sh
+sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t retrieable
+```
+
+### 示例代码
+
+**application.yml**
+
+```yaml
+server:
+  port: 28089
+spring:
+  application:
+    name: rocketmq-retrieable-consume-example
+  cloud:
+    stream:
+      function:
+        definition: consumer;
+      rocketmq:
+        binder:
+          name-server: localhost:9876
+        bindings:
+          producer-out-0:
+            producer:
+              group: output_1
+          consumer-in-0:
+            consumer:
+              ## According to the configured number of `max-reconsume-times`,
+              ## the server will re-push the message according to whether the client's consumption is successful or not
+              push:
+                max-reconsume-times: 3
+      bindings:
+        producer-out-0:
+          destination: retrieable
+        consumer-in-0:
+          destination: retrieable
+          group: retrieable-consumer
+
+logging:
+  level:
+    org.springframework.context.support: debug
+
+```
+
+**code**
+
+```java
+@SpringBootApplication
+public class RocketMQRetrieableConsumeApplication {
+
+    private static final Logger log = LoggerFactory
+        .getLogger(RocketMQRetrieableConsumeApplication.class);
+
+    @Autowired
+    private StreamBridge streamBridge;
+
+    public static void main(String[] args) {
+        SpringApplication.run(RocketMQRetrieableConsumeApplication.class, args);
+    }
+
+    @Bean
+    public ApplicationRunner producer() {
+        return args -> {
+            Map<String, Object> headers = new HashMap<>();
+            Message<SimpleMsg> msg = new GenericMessage(
+                new SimpleMsg("Hello RocketMQ For Retrieable ."), headers);
+            streamBridge.send("producer-out-0", msg);
+        };
+    }
+
+    @Bean
+    public Consumer<Message<SimpleMsg>> consumer() {
+        return msg -> {
+            // Mock Exception in consumer function.
+            throw new RuntimeException("mock exception.");
+        };
+    }
+}
+```
+
 ## Endpoint 信息查看
 
 Spring Boot 应用支持通过 Endpoint 来暴露相关信息，RocketMQ Stream Starter 也支持这一点。

spring-cloud-alibaba-examples/rocketmq-example/readme.md

@@ -807,6 +807,92 @@ public class RocketMQTxApplication {
 }
 ```
 
+## Maximum Retry Consumption Example
+
+- Retry consumption message: According to the configured number of re-consumption, the server will re-push the message according to whether the client's consumption is successful or not.
+
+### Create topic
+
+```sh
+sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t retrieable
+```
+
+### Example code
+
+**application.yml**
+
+```yaml
+server:
+  port: 28089
+spring:
+  application:
+    name: rocketmq-retrieable-consume-example
+  cloud:
+    stream:
+      function:
+        definition: consumer;
+      rocketmq:
+        binder:
+          name-server: localhost:9876
+        bindings:
+          producer-out-0:
+            producer:
+              group: output_1
+          consumer-in-0:
+            consumer:
+              ## According to the configured number of `max-reconsume-times`,
+              ## the server will re-push the message according to whether the client's consumption is successful or not
+              push:
+                max-reconsume-times: 3
+      bindings:
+        producer-out-0:
+          destination: retrieable
+        consumer-in-0:
+          destination: retrieable
+          group: retrieable-consumer
+
+logging:
+  level:
+    org.springframework.context.support: debug
+
+```
+
+**code**
+
+```java
+@SpringBootApplication
+public class RocketMQRetrieableConsumeApplication {
+
+    private static final Logger log = LoggerFactory
+        .getLogger(RocketMQRetrieableConsumeApplication.class);
+
+    @Autowired
+    private StreamBridge streamBridge;
+
+    public static void main(String[] args) {
+        SpringApplication.run(RocketMQRetrieableConsumeApplication.class, args);
+    }
+
+    @Bean
+    public ApplicationRunner producer() {
+        return args -> {
+            Map<String, Object> headers = new HashMap<>();
+            Message<SimpleMsg> msg = new GenericMessage(
+                new SimpleMsg("Hello RocketMQ For Retrieable ."), headers);
+            streamBridge.send("producer-out-0", msg);
+        };
+    }
+
+    @Bean
+    public Consumer<Message<SimpleMsg>> consumer() {
+        return msg -> {
+            // Mock Exception in consumer function.
+            throw new RuntimeException("mock exception.");
+        };
+    }
+}
+```
+
 ## Endpoint
 
 Add dependency `spring-cloud-starter-stream-rocketmq` to your pom.xml file, and configure your endpoint security strategy.

spring-cloud-alibaba-examples/rocketmq-example/rocketmq-retrieable-consume-example/pom.xml

@@ -0,0 +1,57 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+    <parent>
+        <groupId>com.alibaba.cloud</groupId>
+        <artifactId>spring-cloud-alibaba-examples</artifactId>
+        <version>${revision}</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+
+    <artifactId>rocketmq-retrieable-consume-example</artifactId>
+    <name>Spring Cloud Starter Stream Alibaba RocketMQ Retrieable Consume Example</name>
+    <description>Example demonstrating how to use rocketmq to produce, and retrieable consume.</description>
+    <packaging>jar</packaging>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-actuator</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-json</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.cloud</groupId>
+            <artifactId>rocketmq-example-common</artifactId>
+            <version>${revision}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-deploy-plugin</artifactId>
+                <version>${maven-deploy-plugin.version}</version>
+                <configuration>
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

spring-cloud-alibaba-examples/rocketmq-example/rocketmq-retrieable-consume-example/src/main/java/com/alibaba/cloud/examples/retrieable/RocketMQRetrieableConsumeApplication.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 com.alibaba.cloud.examples.retrieable;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import com.alibaba.cloud.examples.common.SimpleMsg;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+/**
+ * RocketMQ Retrieable Consume Example .
+ *
+ * @author Palmer.Xu
+ */
+@SpringBootApplication
+public class RocketMQRetrieableConsumeApplication {
+
+	private static final Logger log = LoggerFactory
+			.getLogger(RocketMQRetrieableConsumeApplication.class);
+
+	@Autowired
+	private StreamBridge streamBridge;
+
+	public static void main(String[] args) {
+		SpringApplication.run(RocketMQRetrieableConsumeApplication.class, args);
+	}
+
+	@Bean
+	public ApplicationRunner producer() {
+		return args -> {
+			Map<String, Object> headers = new HashMap<>();
+			Message<SimpleMsg> msg = new GenericMessage(
+					new SimpleMsg("Hello RocketMQ For Retrieable ."), headers);
+			streamBridge.send("producer-out-0", msg);
+		};
+	}
+
+	@Bean
+	public Consumer<Message<SimpleMsg>> consumer() {
+		return msg -> {
+			throw new RuntimeException("mock exception.");
+		};
+	}
+
+}

spring-cloud-alibaba-examples/rocketmq-example/rocketmq-retrieable-consume-example/src/main/resources/application.yml

@@ -0,0 +1,32 @@
+server:
+  port: 28089
+spring:
+  application:
+    name: rocketmq-retrieable-consume-example
+  cloud:
+    stream:
+      function:
+        definition: consumer;
+      rocketmq:
+        binder:
+          name-server: localhost:9876
+        bindings:
+          producer-out-0:
+            producer:
+              group: output_1
+          consumer-in-0:
+            consumer:
+              ## According to the configured number of `max-reconsume-times`,
+              ## the server will re-push the message according to whether the client's consumption is successful or not
+              push:
+                max-reconsume-times: 3
+      bindings:
+        producer-out-0:
+          destination: retrieable
+        consumer-in-0:
+          destination: retrieable
+          group: retrieable-consumer
+
+logging:
+  level:
+    org.springframework.context.support: debug

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/integration/inbound/RocketMQConsumerFactory.java

@@ -90,6 +90,8 @@ public final class RocketMQConsumerFactory {
 		consumer.setConsumeThreadMin(extendedConsumerProperties.getConcurrency());
 		consumer.setConsumeThreadMax(extendedConsumerProperties.getConcurrency());
 		consumer.setUnitName(consumerProperties.getUnitName());
+		consumer.setMaxReconsumeTimes(
+				consumerProperties.getPush().getMaxReconsumeTimes());
 		return consumer;
 	}