chore: restore retry mechanism for message sending in RocketMQInboundChannelAdapter (#4162)

chore: restore retry mechanism for message sending in RocketMQInboundChannelAdapter (#4162)

Author: xuxiaowei-com-cn

spring-cloud-alibaba-starters/spring-alibaba-nacos-config/src/main/java/com/alibaba/cloud/nacos/constants/Constants.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2023 the original author or authors.
+ * Copyright 2013-present 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.

spring-cloud-alibaba-starters/spring-alibaba-nacos-config/src/test/java/com.alibaba.cloud.nacos/configdata/NacosConfigManagerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2023 the original author or authors.
+ * Copyright 2013-present 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.

spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/pom.xml

@@ -16,6 +16,11 @@
 
         <!--spring boot-->
 
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-webclient</artifactId>
+		</dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-aspectj</artifactId>

spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/webclient/SeataWebClientAutoConfiguration.java

@@ -17,7 +17,7 @@
 package com.alibaba.cloud.seata.webclient;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
-import org.springframework.cloud.client.loadbalancer.reactive.WebClientCustomizer;
+import org.springframework.boot.webclient.WebClientCustomizer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.web.reactive.function.client.WebClient;

spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/webclient/SeataWebClientBuilderCustomizer.java

@@ -16,8 +16,7 @@
 
 package com.alibaba.cloud.seata.webclient;
 
-
-import org.springframework.cloud.client.loadbalancer.reactive.WebClientCustomizer;
+import org.springframework.boot.webclient.WebClientCustomizer;
 import org.springframework.web.reactive.function.client.WebClient;
 
 /**

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

@@ -34,8 +34,11 @@
 import org.slf4j.LoggerFactory;
 
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.core.retry.RetryException;
 import org.springframework.core.retry.RetryListener;
+import org.springframework.core.retry.RetryPolicy;
 import org.springframework.core.retry.RetryTemplate;
+import org.springframework.core.retry.Retryable;
 import org.springframework.integration.context.OrderlyShutdownCapable;
 import org.springframework.integration.core.RecoveryCallback;
 import org.springframework.integration.endpoint.MessageProducerSupport;
@@ -142,10 +145,12 @@ private <R> R consumeMessage(List<MessageExt> messageExtList,
 				Message<?> message = RocketMQMessageConverterSupport
 						.convertMessage2Spring(messageExt);
 				if (this.retryTemplate != null) {
-					// this.retryTemplate.execute(context -> {
-					// 	this.sendMessage(message);
-					// 	return message;
-					// }, this.recoveryCallback);
+					this.retryTemplate.setRetryListener(new RetryListener() {
+						@Override
+						public void onRetryPolicyExhaustion(RetryPolicy retryPolicy, Retryable<?> retryable, RetryException exception) {
+							recoveryCallback.recover(null, exception);
+						}
+					});
 				}
 				else {
 					this.sendMessage(message);

spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/test/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinderTest.java

@@ -23,7 +23,6 @@
 import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
 import jakarta.annotation.Resource;
 import org.assertj.core.api.Assertions;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -35,7 +34,6 @@
 
 import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.NONE;
 
-@Disabled
 @SpringBootTest(classes = RocketMQMessageChannelBinderTest.TestConfig.class,
 	webEnvironment = NONE,
 	properties = {