DATAREDIS-1079 - Fix StreamMessageListenerContainer autoAck on receive.

Original pull request: spring-projects#508.

Author: christophstrobl

src/main/asciidoc/reference/redis-streams.adoc

@@ -163,6 +163,31 @@ Please refer to the Javadoc of the various message listener containers for a ful
 
 NOTE: Demand-driven consumption uses backpressure signals to activate and deactivate polling. `StreamReceiver` subscriptions pause polling if the demand is satisfied until subscribers signal further demand. Depending on the `ReadOffset` strategy, this can cause messages to be skipped.
 
+[[redis.streams.acknowledge]]
+=== `Acknowledge` strategies
+
+When you read with messages via a `Consumer Group`, the server will remember that a given message was delivered and add it to the Pending Entries List (PEL). A list of messages delivered but not yet acknowledged. +
+Messages have to be acknowledged via `StreamOperations.acknowledge` in order to be removed from the Pending Entries List as shown in the snippet below.
+
+====
+[source,java]
+----
+StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = ...
+
+container.receive(Consumer.from("my-group", "my-consumer"), <1>
+StreamOffset.create("my-stream", ReadOffset.lastConsumed()),
+ msg -> {
+
+ // ...
+ redisTemplate.opsForStream().acknowledge("my-group", msg); <2>
+ });
+----
+<1> Read as _my-consumer_ from group _my-group_. Received messages are not acknowledged.
+<2> Acknowledged the message after processing.
+====
+
+TIP: To auto acknowledge messages on receive use `receiveAutoAck` instead of `receive`.
+
 [[redis.streams.receive.readoffset]]
 === `ReadOffset` strategies
 

src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java

@@ -28,6 +28,7 @@
  * Options for reading messages from a Redis Stream.
  * 
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @see 2.2
  */
 @EqualsAndHashCode
@@ -57,14 +58,23 @@ public static StreamReadOptions empty() {
 }
 
 /**
- * Disable auto-acknowledgement when reading in the context of a consumer group.
+ * Enable auto-acknowledgement by setting the {@code NOACK} flag when reading in the context of a consumer group.
  *
  * @return {@link StreamReadOptions} with {@code noack} applied.
  */
 public StreamReadOptions noack() {
 return new StreamReadOptions(block, count, true);
 }
 
+/**
+ * Enable auto-acknowledgement by setting the {@code NOACK} flag when reading in the context of a consumer group.
+ *
+ * @return new instance of {@link StreamReadOptions} with {@code noack} applied.
+ */
+public StreamReadOptions autoAcknowledge() {
+return new StreamReadOptions(block, count, true);
+}
+
 /**
  * Use a blocking read and supply the {@link Duration timeout} after which the call will terminate if no message was
  * read.

src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java

@@ -45,6 +45,7 @@
  * This message container creates long-running tasks that are executed on {@link Executor}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.2
  */
 class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implements StreamMessageListenerContainer<K, V> {
@@ -223,7 +224,7 @@ private StreamPollTask<K, V> getReadTask(StreamReadRequest<K> streamRequest, Str
 
 ConsumerStreamReadRequest<K> consumerStreamRequest = (ConsumerStreamReadRequest<K>) streamRequest;
 
-StreamReadOptions readOptions = consumerStreamRequest.isAutoAck() ? this.readOptions : this.readOptions.noack();
+StreamReadOptions readOptions = consumerStreamRequest.isAutoAck() ? this.readOptions.autoAcknowledge() : this.readOptions;
 Consumer consumer = consumerStreamRequest.getConsumer();
 
 if (this.containerOptions.getHashMapper() != null) {

src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java

@@ -47,6 +47,7 @@
  * Default implementation of {@link StreamReceiver}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.2
  */
 class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver<K, V> {
@@ -134,7 +135,7 @@ public Flux<V> receiveAutoAck(Consumer consumer, StreamOffset<K> streamOffset) {
 }
 
 BiFunction<K, ReadOffset, Flux<? extends Record<?, ?>>> readFunction = getConsumeReadFunction(consumer,
-this.readOptions);
+this.readOptions.autoAcknowledge());
 
 return Flux.defer(() -> {
 
@@ -157,7 +158,7 @@ public Flux<V> receive(Consumer consumer, StreamOffset<K> streamOffset) {
 }
 
 BiFunction<K, ReadOffset, Flux<? extends Record<?, ?>>> readFunction = getConsumeReadFunction(consumer,
-this.readOptions.noack());
+this.readOptions);
 
 return Flux.defer(() -> {
 PollState pollState = PollState.consumer(consumer, streamOffset.getOffset());

src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java

@@ -104,6 +104,7 @@
  * </pre>
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @param <K> Stream key and Stream field type.
  * @param <V> Stream value type.
  * @since 2.2
@@ -196,7 +197,7 @@ default Subscription receive(StreamOffset<K> streamOffset, StreamListener<K, V>
  * @see ReadOffset#lastConsumed()
  */
 default Subscription receive(Consumer consumer, StreamOffset<K> streamOffset, StreamListener<K, V> listener) {
-return register(StreamReadRequest.builder(streamOffset).consumer(consumer).build(), listener);
+return register(StreamReadRequest.builder(streamOffset).consumer(consumer).autoAck(false).build(), listener);
 }
 
 /**

src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java

@@ -18,11 +18,14 @@
 import static org.assertj.core.api.Assertions.*;
 import static org.junit.Assume.*;
 
+import io.lettuce.core.codec.StringCodec;
+import io.lettuce.core.output.NestedMultiOutput;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
 import java.time.Duration;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -38,6 +41,7 @@
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
 import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceConnection;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources;
 import org.springframework.data.redis.connection.stream.Consumer;
@@ -50,11 +54,13 @@
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamMessageListenerContainerOptions;
 import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamReadRequest;
+import org.springframework.util.NumberUtils;
 
 /**
  * Integration tests for {@link StreamMessageListenerContainer}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  */
 public class StreamMessageListenerContainerIntegrationTests {
 
@@ -103,8 +109,7 @@ public void before() {
 public void shouldReceiveMapMessages() throws InterruptedException {
 
 StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
-.create(connectionFactory,
-containerOptions);
+.create(connectionFactory, containerOptions);
 BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
 
 container.start();
@@ -174,12 +179,11 @@ public void shouldReceiveObjectHashRecords() throws InterruptedException {
 assertThat(subscription.isActive()).isFalse();
 }
 
-@Test // DATAREDIS-864
+@Test // DATAREDIS-864, DATAREDIS-1079
 public void shouldReceiveMessagesInConsumerGroup() throws InterruptedException {
 
 StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
-.create(connectionFactory,
-containerOptions);
+.create(connectionFactory, containerOptions);
 BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
 RecordId messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
 redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
@@ -196,6 +200,34 @@ public void shouldReceiveMessagesInConsumerGroup() throws InterruptedException {
 assertThat(message).isNotNull();
 assertThat(message.getValue()).containsEntry("key", "value2");
 
+assertThat(getNumberOfPending("my-stream", "my-group")).isOne();
+
+cancelAwait(subscription);
+}
+
+@Test // DATAREDIS-1079
+public void shouldReceiveAndAckMessagesInConsumerGroup() throws InterruptedException {
+
+StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
+.create(connectionFactory, containerOptions);
+BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
+RecordId messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
+redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
+
+container.start();
+Subscription subscription = container.receiveAutoAck(Consumer.from("my-group", "my-consumer"),
+StreamOffset.create("my-stream", ReadOffset.lastConsumed()), queue::add);
+
+subscription.await(Duration.ofSeconds(2));
+
+redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
+
+MapRecord<String, String, String> message = queue.poll(1, TimeUnit.SECONDS);
+assertThat(message).isNotNull();
+assertThat(message.getValue()).containsEntry("key", "value2");
+
+assertThat(getNumberOfPending("my-stream", "my-group")).isZero();
+
 cancelAwait(subscription);
 }
 
@@ -207,8 +239,7 @@ public void shouldUseCustomErrorHandler() throws InterruptedException {
 StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions = StreamMessageListenerContainerOptions
 .builder().errorHandler(failures::add).pollTimeout(Duration.ofMillis(100)).build();
 StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
-.create(connectionFactory,
-containerOptions);
+.create(connectionFactory, containerOptions);
 
 container.start();
 Subscription subscription = container.receive(Consumer.from("my-group", "my-consumer"),
@@ -229,8 +260,7 @@ public void errorShouldStopListening() throws InterruptedException {
 BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
 
 StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
-.create(connectionFactory,
-containerOptions);
+.create(connectionFactory, containerOptions);
 
 StreamReadRequest<String> readRequest = StreamReadRequest
 .builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())).errorHandler(failures::add)
@@ -260,8 +290,7 @@ public void customizedCancelPredicateShouldNotStopListening() throws Interrupted
 BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
 
 StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
-.create(connectionFactory,
-containerOptions);
+.create(connectionFactory, containerOptions);
 
 StreamReadRequest<String> readRequest = StreamReadRequest
 .builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())) //
@@ -291,8 +320,7 @@ public void customizedCancelPredicateShouldNotStopListening() throws Interrupted
 public void cancelledStreamShouldNotReceiveMessages() throws InterruptedException {
 
 StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
-.create(connectionFactory,
-containerOptions);
+.create(connectionFactory, containerOptions);
 BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
 
 container.start();
@@ -310,8 +338,7 @@ public void cancelledStreamShouldNotReceiveMessages() throws InterruptedExceptio
 public void containerRestartShouldRestartSubscription() throws InterruptedException {
 
 StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
-.create(connectionFactory,
-containerOptions);
+.create(connectionFactory, containerOptions);
 BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
 
 container.start();
@@ -345,6 +372,14 @@ private static void cancelAwait(Subscription subscription) throws InterruptedExc
 }
 }
 
+private Integer getNumberOfPending(String stream, String group) {
+
+String value = ((List) ((LettuceConnection) connectionFactory.getConnection()).execute("XPENDING",
+new NestedMultiOutput(StringCodec.UTF8), new byte[][] { stream.getBytes(), group.getBytes() })).iterator()
+.next().toString();
+return NumberUtils.parseNumber(value, Integer.class);
+}
+
 @Data
 @AllArgsConstructor
 static class LoginEvent {