diff --git a/src/main/asciidoc/reference/redis-streams.adoc b/src/main/asciidoc/reference/redis-streams.adoc index 8f99fbfb0..1e3906bcb 100644 --- a/src/main/asciidoc/reference/redis-streams.adoc +++ b/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> 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 diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java b/src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java index bcb0f4783..814109ecf 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/StreamReadOptions.java +++ b/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. diff --git a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java index c3b8b4090..1faf3033f 100644 --- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java +++ b/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> implements StreamMessageListenerContainer { @@ -223,7 +224,7 @@ class DefaultStreamMessageListenerContainer> implement ConsumerStreamReadRequest consumerStreamRequest = (ConsumerStreamReadRequest) 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) { diff --git a/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java b/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java index ab6a72258..eab382f1a 100644 --- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java +++ b/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> implements StreamReceiver { @@ -134,7 +135,7 @@ class DefaultStreamReceiver> implements StreamReceiver } BiFunction>> readFunction = getConsumeReadFunction(consumer, - this.readOptions); + this.readOptions.autoAcknowledge()); return Flux.defer(() -> { @@ -157,7 +158,7 @@ class DefaultStreamReceiver> implements StreamReceiver } BiFunction>> readFunction = getConsumeReadFunction(consumer, - this.readOptions.noack()); + this.readOptions); return Flux.defer(() -> { PollState pollState = PollState.consumer(consumer, streamOffset.getOffset()); diff --git a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java index 43dd4239c..05d9ec846 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java @@ -104,6 +104,7 @@ import org.springframework.util.ErrorHandler; * * * @author Mark Paluch + * @author Christoph Strobl * @param Stream key and Stream field type. * @param Stream value type. * @since 2.2 @@ -196,7 +197,7 @@ public interface StreamMessageListenerContainer> exten * @see ReadOffset#lastConsumed() */ default Subscription receive(Consumer consumer, StreamOffset streamOffset, StreamListener listener) { - return register(StreamReadRequest.builder(streamOffset).consumer(consumer).build(), listener); + return register(StreamReadRequest.builder(streamOffset).consumer(consumer).autoAck(false).build(), listener); } /** diff --git a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java index c75c322a2..be0c43723 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java +++ b/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> container = StreamMessageListenerContainer - .create(connectionFactory, - containerOptions); + .create(connectionFactory, containerOptions); BlockingQueue> 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> container = StreamMessageListenerContainer - .create(connectionFactory, - containerOptions); + .create(connectionFactory, containerOptions); BlockingQueue> 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> container = StreamMessageListenerContainer + .create(connectionFactory, containerOptions); + BlockingQueue> 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 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> containerOptions = StreamMessageListenerContainerOptions .builder().errorHandler(failures::add).pollTimeout(Duration.ofMillis(100)).build(); StreamMessageListenerContainer> 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 failures = new LinkedBlockingQueue<>(); StreamMessageListenerContainer> container = StreamMessageListenerContainer - .create(connectionFactory, - containerOptions); + .create(connectionFactory, containerOptions); StreamReadRequest readRequest = StreamReadRequest .builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())).errorHandler(failures::add) @@ -260,8 +290,7 @@ public class StreamMessageListenerContainerIntegrationTests { BlockingQueue failures = new LinkedBlockingQueue<>(); StreamMessageListenerContainer> container = StreamMessageListenerContainer - .create(connectionFactory, - containerOptions); + .create(connectionFactory, containerOptions); StreamReadRequest readRequest = StreamReadRequest .builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())) // @@ -291,8 +320,7 @@ public class StreamMessageListenerContainerIntegrationTests { public void cancelledStreamShouldNotReceiveMessages() throws InterruptedException { StreamMessageListenerContainer> container = StreamMessageListenerContainer - .create(connectionFactory, - containerOptions); + .create(connectionFactory, containerOptions); BlockingQueue> queue = new LinkedBlockingQueue<>(); container.start(); @@ -310,8 +338,7 @@ public class StreamMessageListenerContainerIntegrationTests { public void containerRestartShouldRestartSubscription() throws InterruptedException { StreamMessageListenerContainer> container = StreamMessageListenerContainer - .create(connectionFactory, - containerOptions); + .create(connectionFactory, containerOptions); BlockingQueue> 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 {