Fix r2bdc componentType; Redis Streams ack issue

The R2DBC channel adapters have wrong componentType

* Fix R2DBC component type to not mention `reactive-`: they are reactive by R2DBC definition

The `ReactiveRedisStreamMessageProducer` calls ack() explicitly
when populates `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header,
but must provide an implementation of `SimpleAcknowledgment` instead

* Fix `ReactiveRedisStreamMessageProducer` to populate the proper
`IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header and in the proper moment
* Fix `ReactiveRedisStreamMessageProducerTests` to verify that `autoAck=false` works as expected
together with the `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header functionality

Author: rohan mukesh

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java

@@ -112,7 +112,7 @@ public void setCriteriaExpression(Expression criteriaExpression) {
 
 	@Override
 	public String getComponentType() {
-		return "r2dbc:reactive-outbound-channel-adapter";
+		return "r2dbc:outbound-channel-adapter";
 	}
 
 	@Override

spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java

@@ -89,6 +89,12 @@ public void setup() {
 				.verifyComplete());
 	}
 
+	@Test
+	public void validateComponentType() {
+		assertThat(this.defaultR2dbcMessageSource.getComponentType())
+												.isEqualTo("r2dbc:inbound-channel-adapter");
+	}
+
 	@Test
 	public void validateSuccessfulQueryWithoutSettingExpectedElement() {
 		this.entityTemplate.insert(new Person("Bob", 35))

spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandlerTests.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.r2dbc.outbound;
 
+import static org.assertj.core.api.Assertions.assertThat;
 
 import java.time.Duration;
 import java.util.Arrays;
@@ -115,6 +116,11 @@ public void validateMessageHandlingWithInsertQueryCollection() {
 
 	}
 
+	@Test
+	public void validateComponentType() {
+		assertThat(this.r2dbcMessageHandler.getComponentType()).isEqualTo("r2dbc:outbound-channel-adapter");
+	}
+
 	@Test
 	public void validateMessageHandlingWithDefaultUpdateCollection() {
 		Message<Person> message = MessageBuilder.withPayload(createPerson("Bob", 35)).build();

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java

@@ -28,6 +28,7 @@
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.data.redis.stream.StreamReceiver;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.acks.SimpleAcknowledgment;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.redis.support.RedisHeaders;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
@@ -48,6 +49,7 @@
  *
  * @author Attoumane Ahamadi
  * @author Artem Bilan
+ * @author Rohan Mukesh
  *
  * @since 5.4
  */
@@ -217,10 +219,12 @@ protected void doStart() {
 									.setHeader(RedisHeaders.STREAM_MESSAGE_ID, event.getId())
 									.setHeader(RedisHeaders.CONSUMER_GROUP, this.consumerGroup)
 									.setHeader(RedisHeaders.CONSUMER, this.consumerName);
-					if (!this.autoAck) {
+					if (!this.autoAck && this.consumerGroup != null) {
 						builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
-								this.reactiveStreamOperations.acknowledge(this.consumerGroup, event)
-										.subscribe());
+								(SimpleAcknowledgment) () ->
+										this.reactiveStreamOperations
+												.acknowledge(this.consumerGroup, event)
+												.subscribe());
 					}
 					return builder.build();
 				});

spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java

@@ -19,6 +19,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.time.Duration;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.After;
 import org.junit.Before;
@@ -28,10 +29,15 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
 import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamInfo;
 import org.springframework.data.redis.core.ReactiveRedisTemplate;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.data.redis.stream.StreamReceiver;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.StaticMessageHeaderAccessor;
+import org.springframework.integration.acks.SimpleAcknowledgment;
 import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.redis.outbound.ReactiveRedisStreamMessageHandler;
@@ -46,11 +52,13 @@
 import org.springframework.test.context.junit4.SpringRunner;
 
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 /**
  * @author Attoumane Ahamadi
  * @author Artem Bilan
+ * @author Rohan Mukesh
  *
  * @since 5.4
  */
@@ -76,6 +84,16 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
 
 	@Before
 	public void delKey() {
+		this.template.hasKey(STREAM_KEY)
+				.filter(Boolean::booleanValue)
+				.flatMapMany(b ->
+						this.template.opsForStream()
+								.groups(STREAM_KEY)
+								.map(StreamInfo.XInfoGroup::groupName)
+								.flatMap(groupName ->
+										this.template.opsForStream()
+												.destroyGroup(STREAM_KEY, groupName)))
+				.blockLast();
 		this.template.delete(STREAM_KEY).block();
 	}
 
@@ -90,10 +108,11 @@ public void testConsumerGroupCreation() {
 		this.redisStreamMessageProducer.setCreateConsumerGroup(true);
 		this.redisStreamMessageProducer.setConsumerName(CONSUMER);
 		this.redisStreamMessageProducer.afterPropertiesSet();
-		this.redisStreamMessageProducer.start();
 
 		Flux.from(this.fluxMessageChannel).subscribe();
 
+		this.redisStreamMessageProducer.start();
+
 		this.template.opsForStream()
 				.groups(STREAM_KEY)
 				.next()
@@ -147,19 +166,81 @@ public void testReadingMessageAsConsumerInConsumerGroup() {
 
 		this.redisStreamMessageProducer.setCreateConsumerGroup(false);
 		this.redisStreamMessageProducer.setConsumerName(CONSUMER);
+		this.redisStreamMessageProducer.setReadOffset(ReadOffset.latest());
 		this.redisStreamMessageProducer.afterPropertiesSet();
 		this.redisStreamMessageProducer.start();
 
+		StepVerifier stepVerifier =
+				Flux.from(this.fluxMessageChannel)
+						.as(StepVerifier::create)
+						.assertNext(message -> {
+							assertThat(message.getPayload()).isEqualTo(person);
+							assertThat(message.getHeaders()).containsKeys(RedisHeaders.CONSUMER_GROUP, RedisHeaders.CONSUMER);
+						})
+						.thenCancel()
+						.verifyLater();
+
 		this.messageHandler.handleMessage(new GenericMessage<>(person));
 
-		Flux.from(this.fluxMessageChannel)
+		stepVerifier.verify(Duration.ofSeconds(10));
+	}
+
+	@Test
+	@RedisAvailable
+	public void testReadingPendingMessageWithNoAutoACK() {
+		Address address = new Address("Winterfell, Westeros");
+		Person person = new Person(address, "John Snow");
+
+		this.template.opsForStream()
+				.createGroup(STREAM_KEY, this.redisStreamMessageProducer.getBeanName())
 				.as(StepVerifier::create)
-				.assertNext(message -> {
-					assertThat(message.getPayload()).isEqualTo(person);
-					assertThat(message.getHeaders()).containsKeys(RedisHeaders.CONSUMER_GROUP, RedisHeaders.CONSUMER);
-				})
+				.assertNext(message -> assertThat(message).isEqualTo("OK"))
 				.thenCancel()
 				.verify(Duration.ofSeconds(10));
+
+		this.redisStreamMessageProducer.setCreateConsumerGroup(false);
+		this.redisStreamMessageProducer.setAutoAck(false);
+		this.redisStreamMessageProducer.setConsumerName(CONSUMER);
+		this.redisStreamMessageProducer.setReadOffset(ReadOffset.latest());
+		this.redisStreamMessageProducer.afterPropertiesSet();
+		this.redisStreamMessageProducer.start();
+
+		AtomicReference<SimpleAcknowledgment> acknowledgmentReference = new AtomicReference<>();
+
+		StepVerifier stepVerifier =
+				Flux.from(this.fluxMessageChannel)
+						.as(StepVerifier::create)
+						.assertNext(message -> {
+							assertThat(message.getPayload()).isEqualTo(person);
+							assertThat(message.getHeaders())
+									.containsKeys(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK);
+							acknowledgmentReference.set(StaticMessageHeaderAccessor.getAcknowledgment(message));
+						})
+						.thenCancel()
+						.verifyLater();
+
+		this.messageHandler.handleMessage(new GenericMessage<>(person));
+
+		stepVerifier.verify(Duration.ofSeconds(10));
+
+		Mono<PendingMessagesSummary> pending =
+				template.opsForStream()
+						.pending(STREAM_KEY, this.redisStreamMessageProducer.getBeanName());
+
+		StepVerifier.create(pending)
+				.assertNext(pendingMessagesSummary ->
+						assertThat(pendingMessagesSummary.getTotalPendingMessages()).isEqualTo(1))
+				.verifyComplete();
+
+		acknowledgmentReference.get().acknowledge();
+
+		Mono<PendingMessagesSummary> pendingZeroMessage = template.opsForStream().pending(STREAM_KEY,
+				this.redisStreamMessageProducer.getBeanName());
+
+		StepVerifier.create(pendingZeroMessage)
+				.assertNext(pendingMessagesSummary ->
+						assertThat(pendingMessagesSummary.getTotalPendingMessages()).isEqualTo(0))
+				.verifyComplete();
 	}
 
 	@Configuration