DATAREDIS-1079 - Fix StreamMessageListenerContainer autoAck on receive.

Original pull request: #508.

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 @@ import org.springframework.util.Assert;
  * Options for reading messages from a Redis Stream.
  * 
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @see 2.2
  */
 @EqualsAndHashCode
@@ -57,7 +58,7 @@ public class StreamReadOptions {
 	}
 
 	/**
-	 * 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.
 	 */
@@ -65,6 +66,15 @@ public class StreamReadOptions {
 		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 @@ import org.springframework.util.ErrorHandler;
  * 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 @@ class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implement
 
 			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 @@ import org.springframework.data.redis.serializer.RedisSerializationContext;
  * 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 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
 		}
 
 		BiFunction<K, ReadOffset, Flux<? extends Record<?, ?>>> readFunction = getConsumeReadFunction(consumer,
-				this.readOptions);
+				this.readOptions.autoAcknowledge());
 
 		return Flux.defer(() -> {
 
@@ -157,7 +158,7 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
 		}
 
 		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 @@ import org.springframework.util.ErrorHandler;
  * </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 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 	 * @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 @@ package org.springframework.data.redis.stream;
 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.RedisConnection;
 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.connection.stream.StreamOffset;
 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 class StreamMessageListenerContainerIntegrationTests {
 	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 class StreamMessageListenerContainerIntegrationTests {
 		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 class StreamMessageListenerContainerIntegrationTests {
 		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 class StreamMessageListenerContainerIntegrationTests {
 		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 class StreamMessageListenerContainerIntegrationTests {
 		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 class StreamMessageListenerContainerIntegrationTests {
 		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 class StreamMessageListenerContainerIntegrationTests {
 	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 class StreamMessageListenerContainerIntegrationTests {
 	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 @@ public class StreamMessageListenerContainerIntegrationTests {
 		}
 	}
 
+	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 {