From a5700ba71a4bcccfa1a561e1a2d2c123ac14d4d5 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Tue, 24 Nov 2020 15:09:09 +0100 Subject: [PATCH] DATAREDIS-1230 - Revise error handling in StreamReceiver and StreamMessageListenerContainer. Reading and deserialization in StreamReceiver and StreamMessageListener is now decoupled from each other to allow fine-grained control over errors and resumption. Previously, we used the Template API to read and deserialize stream records. Now the actual read happens before the deserialization so that errors during deserialization of individual messages can be handled properly. This split also allows advancing in the stream read. Previously, the failed deserialization prevented of getting hold of the non-serialized Stream record which caused the stream receiver to remain at the offset that fetched the offending record which effectively lead to an infinite loop. We also support a resume function in StreamReceiver to control whether stream reads should be resumed or terminated. Original Pull Request: #576 --- .../core/DefaultReactiveStreamOperations.java | 3 +- .../redis/core/DefaultStreamOperations.java | 7 +- .../redis/core/ReactiveStreamOperations.java | 47 +++++--- .../data/redis/core/StreamObjectMapper.java | 10 +- .../data/redis/core/StreamOperations.java | 62 ++++++++--- ...DefaultStreamMessageListenerContainer.java | 59 ++++++---- .../redis/stream/DefaultStreamReceiver.java | 104 +++++++++++++----- .../StreamMessageListenerContainer.java | 21 +++- .../data/redis/stream/StreamPollTask.java | 66 +++++++++-- .../data/redis/stream/StreamReceiver.java | 58 +++++++++- ...sageListenerContainerIntegrationTests.java | 50 ++++++++- .../StreamReceiverIntegrationTests.java | 36 ++++++ 12 files changed, 415 insertions(+), 108 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java index 632c72299..b7e907ea6 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java @@ -416,7 +416,8 @@ class DefaultReactiveStreamOperations implements ReactiveStreamOperat return (HV) serializationContext.getHashValueSerializationPair().read(buffer); } - private MapRecord deserializeRecord(ByteBufferRecord record) { + @Override + public MapRecord deserializeRecord(ByteBufferRecord record) { return record.map(it -> it.mapEntries(this::deserializeRecordFields).withStreamKey(readKey(record.getStream()))); } diff --git a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java index e59cf85c7..4db28ee49 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java @@ -347,6 +347,11 @@ class DefaultStreamOperations extends AbstractOperations i return objectMapper.getHashMapper(targetType); } + @Override + public MapRecord deserializeRecord(ByteRecord record) { + return record.deserialize(keySerializer(), hashKeySerializer(), hashValueSerializer()); + } + protected byte[] serializeHashValueIfRequires(HV value) { return hashValueSerializerPresent() ? serialize(value, hashValueSerializer()) : objectMapper.getConversionService().convert(value, byte[].class); @@ -386,7 +391,7 @@ class DefaultStreamOperations extends AbstractOperations i List> result = new ArrayList<>(); for (ByteRecord record : raw) { - result.add(record.deserialize(keySerializer(), hashKeySerializer(), hashValueSerializer())); + result.add(deserializeRecord(record)); } return result; diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java index 0892fa1e1..e78437c0a 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java @@ -22,23 +22,13 @@ import java.util.Arrays; import java.util.Map; import org.reactivestreams.Publisher; + import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; -import org.springframework.data.redis.connection.stream.Consumer; -import org.springframework.data.redis.connection.stream.MapRecord; -import org.springframework.data.redis.connection.stream.ObjectRecord; -import org.springframework.data.redis.connection.stream.PendingMessage; -import org.springframework.data.redis.connection.stream.PendingMessages; -import org.springframework.data.redis.connection.stream.PendingMessagesSummary; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.Record; -import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.*; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroup; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream; -import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.connection.stream.StreamReadOptions; -import org.springframework.data.redis.connection.stream.StreamRecords; import org.springframework.data.redis.hash.HashMapper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -354,7 +344,7 @@ public interface ReactiveStreamOperations extends HashMapperProvider< Assert.notNull(targetType, "Target type must not be null"); - return range(key, range, limit).map(it -> StreamObjectMapper.toObjectRecord(this, it, targetType)); + return range(key, range, limit).map(it -> map(it, targetType)); } /** @@ -435,7 +425,7 @@ public interface ReactiveStreamOperations extends HashMapperProvider< Assert.notNull(targetType, "Target type must not be null"); - return read(readOptions, streams).map(it -> StreamObjectMapper.toObjectRecord(this, it, targetType)); + return read(readOptions, streams).map(it -> map(it, targetType)); } /** @@ -489,7 +479,7 @@ public interface ReactiveStreamOperations extends HashMapperProvider< Assert.notNull(targetType, "Target type must not be null"); - return read(consumer, readOptions, streams).map(it -> StreamObjectMapper.toObjectRecord(this, it, targetType)); + return read(consumer, readOptions, streams).map(it -> map(it, targetType)); } /** @@ -543,7 +533,7 @@ public interface ReactiveStreamOperations extends HashMapperProvider< Assert.notNull(targetType, "Target type must not be null"); - return reverseRange(key, range, limit).map(it -> StreamObjectMapper.toObjectRecord(this, it, targetType)); + return reverseRange(key, range, limit).map(it -> map(it, targetType)); } /** @@ -577,4 +567,29 @@ public interface ReactiveStreamOperations extends HashMapperProvider< */ @Override HashMapper getHashMapper(Class targetType); + + /** + * Map records from {@link MapRecord} to {@link ObjectRecord}. + * + * @param record the stream records to map. + * @param targetType the target type of the payload. + * @return the mapped {@link ObjectRecord}. + * @since 2.x + */ + default ObjectRecord map(MapRecord record, Class targetType) { + + Assert.notNull(record, "Records must not be null"); + Assert.notNull(targetType, "Target type must not be null"); + + return StreamObjectMapper.toObjectRecord(record, this, targetType); + } + + /** + * Deserialize a {@link ByteBufferRecord} using the configured serialization context into a {@link MapRecord}. + * + * @param record the stream record to map. + * @return deserialized {@link MapRecord}. + * @since 2.x + */ + MapRecord deserializeRecord(ByteBufferRecord record); } diff --git a/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java b/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java index 87604f2cc..7c7067492 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java +++ b/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java @@ -128,13 +128,13 @@ class StreamObjectMapper { /** * Convert the given {@link Record} into an {@link ObjectRecord}. * - * @param provider provider for {@link HashMapper} to apply mapping for {@link ObjectRecord}. * @param source the source value. + * @param provider provider for {@link HashMapper} to apply mapping for {@link ObjectRecord}. * @param targetType the desired target type. * @return the converted {@link ObjectRecord}. */ - static ObjectRecord toObjectRecord(HashMapperProvider provider, - MapRecord source, Class targetType) { + static ObjectRecord toObjectRecord(MapRecord source, + HashMapperProvider provider, Class targetType) { return source.toObjectRecord(provider.getHashMapper(targetType)); } @@ -149,7 +149,7 @@ class StreamObjectMapper { * {@literal null}. */ @Nullable - static List> map(@Nullable List> records, + static List> toObjectRecords(@Nullable List> records, HashMapperProvider hashMapperProvider, Class targetType) { if (records == null) { @@ -161,7 +161,7 @@ class StreamObjectMapper { } if (records.size() == 1) { - return Collections.singletonList(toObjectRecord(hashMapperProvider, records.get(0), targetType)); + return Collections.singletonList(toObjectRecord(records.get(0), hashMapperProvider, targetType)); } List> transformed = new ArrayList<>(records.size()); diff --git a/src/main/java/org/springframework/data/redis/core/StreamOperations.java b/src/main/java/org/springframework/data/redis/core/StreamOperations.java index 41537daca..776a9d863 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -23,21 +23,10 @@ import java.util.Map; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; -import org.springframework.data.redis.connection.stream.Consumer; -import org.springframework.data.redis.connection.stream.MapRecord; -import org.springframework.data.redis.connection.stream.ObjectRecord; -import org.springframework.data.redis.connection.stream.PendingMessage; -import org.springframework.data.redis.connection.stream.PendingMessages; -import org.springframework.data.redis.connection.stream.PendingMessagesSummary; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.Record; -import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.*; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumers; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroups; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream; -import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.connection.stream.StreamReadOptions; -import org.springframework.data.redis.connection.stream.StreamRecords; import org.springframework.data.redis.hash.HashMapper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -356,7 +345,7 @@ public interface StreamOperations extends HashMapperProvider Assert.notNull(targetType, "Target type must not be null"); - return StreamObjectMapper.map(range(key, range, limit), this, targetType); + return map(range(key, range, limit), targetType); } /** @@ -409,7 +398,7 @@ public interface StreamOperations extends HashMapperProvider Assert.notNull(targetType, "Target type must not be null"); - return StreamObjectMapper.map(read(readOptions, streams), this, targetType); + return map(read(readOptions, streams), targetType); } /** @@ -467,7 +456,7 @@ public interface StreamOperations extends HashMapperProvider Assert.notNull(targetType, "Target type must not be null"); - return StreamObjectMapper.map(read(consumer, readOptions, streams), this, targetType); + return map(read(consumer, readOptions, streams), targetType); } /** @@ -523,7 +512,7 @@ public interface StreamOperations extends HashMapperProvider Assert.notNull(targetType, "Target type must not be null"); - return StreamObjectMapper.map(reverseRange(key, range, limit), this, targetType); + return map(reverseRange(key, range, limit), targetType); } /** @@ -560,4 +549,45 @@ public interface StreamOperations extends HashMapperProvider @Override HashMapper getHashMapper(Class targetType); + /** + * Map record from {@link MapRecord} to {@link ObjectRecord}. + * + * @param record the stream record to map. + * @param targetType the target type of the payload. + * @return the mapped {@link ObjectRecord}. + * @since 2.x + */ + default ObjectRecord map(MapRecord record, Class targetType) { + + Assert.notNull(record, "Record must not be null"); + Assert.notNull(targetType, "Target type must not be null"); + + return StreamObjectMapper.toObjectRecord(record, this, targetType); + } + + /** + * Map records from {@link MapRecord} to {@link ObjectRecord}s. + * + * @param records the stream records to map. + * @param targetType the target type of the payload. + * @return the mapped {@link ObjectRecord object records}. + * @since 2.x + */ + @Nullable + default List> map(@Nullable List> records, Class targetType) { + + Assert.notNull(records, "Records must not be null"); + Assert.notNull(targetType, "Target type must not be null"); + + return StreamObjectMapper.toObjectRecords(records, this, targetType); + } + + /** + * Deserialize a {@link ByteRecord} using the configured serializers into a {@link MapRecord}. + * + * @param record the stream record to map. + * @return deserialized {@link MapRecord}. + * @since 2.x + */ + MapRecord deserializeRecord(ByteRecord record); } 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 fce031f4b..b5b372909 100644 --- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java @@ -19,19 +19,25 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Executor; -import java.util.function.BiFunction; +import java.util.function.Function; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + +import org.springframework.core.convert.TypeDescriptor; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.Record; import org.springframework.data.redis.connection.stream.StreamOffset; import org.springframework.data.redis.connection.stream.StreamReadOptions; +import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StreamOperations; +import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; import org.springframework.util.ObjectUtils; @@ -79,8 +85,8 @@ class DefaultStreamMessageListenerContainer> implement this.template = createRedisTemplate(connectionFactory, containerOptions); this.containerOptions = containerOptions; - if (containerOptions.getHashMapper() != null) { - this.streamOperations = this.template.opsForStream(containerOptions.getHashMapper()); + if (containerOptions.hasHashMapper()) { + this.streamOperations = this.template.opsForStream(containerOptions.getRequiredHashMapper()); } else { this.streamOperations = this.template.opsForStream(); } @@ -207,16 +213,39 @@ class DefaultStreamMessageListenerContainer> implement return doRegister(getReadTask(streamRequest, listener)); } - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) private StreamPollTask getReadTask(StreamReadRequest streamRequest, StreamListener listener) { - BiFunction>> readFunction = getReadFunction(streamRequest); + Function> readFunction = getReadFunction(streamRequest); + Function deserializerToUse = getDeserializer(); - return new StreamPollTask<>(streamRequest, listener, errorHandler, (BiFunction) readFunction); + TypeDescriptor targetType = TypeDescriptor + .valueOf(containerOptions.hasHashMapper() ? containerOptions.getTargetType() : MapRecord.class); + + return new StreamPollTask<>(streamRequest, listener, errorHandler, targetType, readFunction, deserializerToUse); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + private Function getDeserializer() { + + Function> deserializer = streamOperations::deserializeRecord; + + if (containerOptions.getHashMapper() == null) { + return (Function) deserializer; + } + + return source -> { + + MapRecord intermediate = deserializer.apply(source); + return (V) streamOperations.map(intermediate, this.containerOptions.getTargetType()); + }; } @SuppressWarnings("unchecked") - private BiFunction>> getReadFunction(StreamReadRequest streamRequest) { + private Function> getReadFunction(StreamReadRequest streamRequest) { + + byte[] rawKey = ((RedisSerializer) template.getKeySerializer()) + .serialize(streamRequest.getStreamOffset().getKey()); if (streamRequest instanceof StreamMessageListenerContainer.ConsumerStreamReadRequest) { @@ -226,20 +255,12 @@ class DefaultStreamMessageListenerContainer> implement : this.readOptions; Consumer consumer = consumerStreamRequest.getConsumer(); - if (this.containerOptions.getHashMapper() != null) { - return (key, offset) -> streamOperations.read(this.containerOptions.getTargetType(), consumer, readOptions, - StreamOffset.create(key, offset)); - } - - return (key, offset) -> streamOperations.read(consumer, readOptions, StreamOffset.create(key, offset)); + return (offset) -> template.execute((RedisCallback>) connection -> connection.streamCommands() + .xReadGroup(consumer, readOptions, StreamOffset.create(rawKey, offset))); } - if (this.containerOptions.getHashMapper() != null) { - return (key, offset) -> streamOperations.read(this.containerOptions.getTargetType(), readOptions, - StreamOffset.create(key, offset)); - } - - return (key, offset) -> streamOperations.read(readOptions, StreamOffset.create(key, offset)); + return (offset) -> template.execute((RedisCallback>) connection -> connection.streamCommands() + .xRead(readOptions, StreamOffset.create(rawKey, offset))); } private Subscription doRegister(Task task) { 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 7d2c17159..0eff3f021 100644 --- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java +++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java @@ -22,18 +22,25 @@ import reactor.core.publisher.Operators; import reactor.util.concurrent.Queues; import reactor.util.context.Context; +import java.nio.ByteBuffer; import java.util.Optional; import java.util.Queue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; -import java.util.function.BiFunction; +import java.util.function.Function; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.reactivestreams.Publisher; import org.reactivestreams.Subscription; + +import org.springframework.core.convert.ConversionFailedException; +import org.springframework.core.convert.TypeDescriptor; import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; +import org.springframework.data.redis.connection.stream.ByteBufferRecord; import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.Record; import org.springframework.data.redis.connection.stream.StreamOffset; @@ -66,6 +73,7 @@ class DefaultStreamReceiver> implements StreamReceiver */ @SuppressWarnings("unchecked") DefaultStreamReceiver(ReactiveRedisConnectionFactory connectionFactory, StreamReceiverOptions options) { + receiverOptions = options; RedisSerializationContext serializationContext = RedisSerializationContext @@ -85,8 +93,8 @@ class DefaultStreamReceiver> implements StreamReceiver this.readOptions = readOptions; this.template = new ReactiveRedisTemplate(connectionFactory, serializationContext); - if (options.getHashMapper() != null) { - this.streamOperations = this.template.opsForStream(options.getHashMapper()); + if (options.hasHashMapper()) { + this.streamOperations = this.template.opsForStream(options.getRequiredHashMapper()); } else { this.streamOperations = this.template.opsForStream(); } @@ -104,20 +112,19 @@ class DefaultStreamReceiver> implements StreamReceiver logger.debug(String.format("receive(%s)", streamOffset)); } - BiFunction>> readFunction; + RedisSerializationContext.SerializationPair keySerializer = template.getSerializationContext() + .getKeySerializationPair(); + ByteBuffer rawKey = keySerializer.write(streamOffset.getKey()); - if (receiverOptions.getHashMapper() != null) { - readFunction = (key, readOffset) -> streamOperations.read(receiverOptions.getTargetType(), readOptions, - StreamOffset.create(key, readOffset)); - } else { - readFunction = (key, readOffset) -> streamOperations.read(readOptions, StreamOffset.create(key, readOffset)); - } + Function> readFunction = readOffset -> template.execute(connection -> connection + .streamCommands().xRead(readOptions, StreamOffset.create(rawKey.asReadOnlyBuffer(), readOffset))); return Flux.defer(() -> { PollState pollState = PollState.standalone(streamOffset.getOffset()); return Flux.create( - sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, (BiFunction) readFunction).arm()); + sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, readFunction, + receiverOptions.getResumeFunction()).arm()); }); } @@ -133,14 +140,16 @@ class DefaultStreamReceiver> implements StreamReceiver logger.debug(String.format("receiveAutoAck(%s, %s)", consumer, streamOffset)); } - BiFunction>> readFunction = getConsumeReadFunction(consumer, + Function> readFunction = getConsumeReadFunction(streamOffset.getKey(), consumer, this.readOptions.autoAcknowledge()); + return Flux.defer(() -> { PollState pollState = PollState.consumer(consumer, streamOffset.getOffset()); return Flux.create( - sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, (BiFunction) readFunction).arm()); + sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, readFunction, + receiverOptions.getResumeFunction()).arm()); }); } @@ -156,26 +165,43 @@ class DefaultStreamReceiver> implements StreamReceiver logger.debug(String.format("receive(%s, %s)", consumer, streamOffset)); } - BiFunction>> readFunction = getConsumeReadFunction(consumer, + Function> readFunction = getConsumeReadFunction(streamOffset.getKey(), consumer, this.readOptions); return Flux.defer(() -> { PollState pollState = PollState.consumer(consumer, streamOffset.getOffset()); return Flux.create( - sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, (BiFunction) readFunction).arm()); + sink -> new StreamSubscription(sink, streamOffset.getKey(), pollState, readFunction, + receiverOptions.getResumeFunction()).arm()); }); } @SuppressWarnings("unchecked") - private BiFunction>> getConsumeReadFunction(Consumer consumer, + private Function> getConsumeReadFunction(K key, Consumer consumer, StreamReadOptions readOptions) { - if (receiverOptions.getHashMapper() != null) { - return (key, readOffset) -> streamOperations.read(receiverOptions.getTargetType(), consumer, readOptions, - StreamOffset.create(key, readOffset)); + RedisSerializationContext.SerializationPair keySerializer = template.getSerializationContext() + .getKeySerializationPair(); + ByteBuffer rawKey = keySerializer.write(key); + + return readOffset -> template.execute(connection -> connection.streamCommands().xReadGroup(consumer, readOptions, + StreamOffset.create(rawKey.asReadOnlyBuffer(), readOffset))); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + private Function getDeserializer() { + + Function> deserializer = streamOperations::deserializeRecord; + + if (receiverOptions.getHashMapper() == null) { + return (Function) deserializer; } - return (key, readOffset) -> streamOperations.read(consumer, readOptions, StreamOffset.create(key, readOffset)); + return source -> { + + MapRecord intermediate = deserializer.apply(source); + return (V) streamOperations.map(intermediate, this.receiverOptions.getTargetType()); + }; } /** @@ -188,15 +214,23 @@ class DefaultStreamReceiver> implements StreamReceiver private final FluxSink sink; private final K key; private final PollState pollState; - private final BiFunction> readFunction; + private final Function> readFunction; + private final Function> resumeFunction; + private final Function deserializer; + private final TypeDescriptor targetType; protected StreamSubscription(FluxSink sink, K key, PollState pollState, - BiFunction> readFunction) { + Function> readFunction, + Function> resumeFunction) { this.sink = sink; this.key = key; this.pollState = pollState; this.readFunction = readFunction; + this.resumeFunction = resumeFunction; + this.deserializer = getDeserializer(); + this.targetType = TypeDescriptor + .valueOf(receiverOptions.hasHashMapper() ? receiverOptions.getTargetType() : MapRecord.class); } /** @@ -237,6 +271,7 @@ class DefaultStreamReceiver> implements StreamReceiver sink.onCancel(pollState::cancel); } + @SuppressWarnings({ "unchecked", "ConstantConditions" }) private void scheduleIfRequired() { if (logger.isDebugEnabled()) { @@ -290,9 +325,24 @@ class DefaultStreamReceiver> implements StreamReceiver String.format("[stream: %s] scheduleIfRequired(): Activating subscription, offset %s", key, readOffset)); } - Flux poll = readFunction.apply(key, readOffset); + Flux poll = readFunction.apply(readOffset) + .onErrorResume(throwable -> Flux.from(resumeFunction.apply(throwable)).then().cast(ByteBufferRecord.class)); - poll.subscribe(getSubscriber()); + poll.map(it -> { + + if (logger.isDebugEnabled()) { + logger.debug(String.format("[stream: %s] onStreamMessage(%s)", key, it)); + } + + pollState.updateReadOffset(it.getId().getValue()); + + try { + return deserializer.apply(it); + } catch (RuntimeException e) { + throw new ConversionFailedException(TypeDescriptor.forObject(it), targetType, it, e); + } + }).onErrorResume(throwable -> Flux.from(resumeFunction.apply(throwable)).then().map(it -> (V) new Object())) // + .subscribe(getSubscriber()); } } @@ -336,12 +386,6 @@ class DefaultStreamReceiver> implements StreamReceiver private void onStreamMessage(V message) { - if (logger.isDebugEnabled()) { - logger.debug(String.format("[stream: %s] onStreamMessage(%s)", key, message)); - } - - pollState.updateReadOffset(message.getId().getValue()); - long requested = pollState.getRequested(); if (requested > 0) { 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 4c17aad5d..75550d997 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java @@ -417,7 +417,7 @@ public interface StreamMessageListenerContainer> exten } /** - * Configure a {@link ErrorHandler} to be notified on {@link Throwable errors}. + * Configure a {@link ErrorHandler} to be notified on {@link Throwable read, deserialization, and listener errors}. * * @param errorHandler must not be null. * @return {@code this} {@link ConsumerStreamReadRequestBuilder}. @@ -429,8 +429,9 @@ public interface StreamMessageListenerContainer> exten } /** - * Configure a cancellation {@link Predicate} to be notified on {@link Throwable errors}. The outcome of the - * {@link Predicate} decides whether to cancel the subscription by returning {@literal true}. + * Configure a cancellation {@link Predicate} to be notified on {@link Throwable read, deserialization, and listener + * errors}. The outcome of the {@link Predicate} decides whether to cancel the subscription by returning + * {@literal true}. * * @param cancelSubscriptionOnError must not be null. * @return {@code this} {@link ConsumerStreamReadRequestBuilder}. @@ -568,6 +569,19 @@ public interface StreamMessageListenerContainer> exten return hashMapper; } + public HashMapper getRequiredHashMapper() { + + if (!hasHashMapper()) { + throw new IllegalStateException("No HashMapper configured"); + } + + return hashMapper; + } + + public boolean hasHashMapper() { + return hashMapper != null; + } + public Class getTargetType() { if (this.targetType != null) { @@ -590,6 +604,7 @@ public interface StreamMessageListenerContainer> exten public Executor getExecutor() { return executor; } + } /** diff --git a/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java b/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java index e4ed51ad9..d78a6ab5b 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java @@ -21,9 +21,13 @@ import java.util.Optional; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; +import java.util.function.Function; import java.util.function.Predicate; +import org.springframework.core.convert.ConversionFailedException; +import org.springframework.core.convert.TypeDescriptor; import org.springframework.dao.DataAccessResourceFailureException; +import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.Record; @@ -32,6 +36,7 @@ import org.springframework.data.redis.stream.StreamMessageListenerContainer.Cons import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamReadRequest; import org.springframework.util.ErrorHandler; + /** * {@link Task} that invokes a {@link BiFunction read function} to poll on a Redis Stream. * @@ -40,24 +45,28 @@ import org.springframework.util.ErrorHandler; */ class StreamPollTask> implements Task { - private final StreamReadRequest request; private final StreamListener listener; private final ErrorHandler errorHandler; private final Predicate cancelSubscriptionOnError; - private final BiFunction> readFunction; + private final Function> readFunction; + private final Function deserializer; private final PollState pollState; + private final TypeDescriptor targetType; + private volatile boolean isInEventLoop = false; StreamPollTask(StreamReadRequest streamRequest, StreamListener listener, ErrorHandler errorHandler, - BiFunction> readFunction) { + TypeDescriptor targetType, Function> readFunction, + Function deserializer) { - this.request = streamRequest; this.listener = listener; this.errorHandler = Optional.ofNullable(streamRequest.getErrorHandler()).orElse(errorHandler); this.cancelSubscriptionOnError = streamRequest.getCancelSubscriptionOnError(); this.readFunction = readFunction; + this.deserializer = deserializer; this.pollState = createPollState(streamRequest); + this.targetType = targetType; } private static PollState createPollState(StreamReadRequest streamRequest) { @@ -120,13 +129,13 @@ class StreamPollTask> implements Task { isInEventLoop = true; pollState.running(); - doLoop(request.getStreamOffset().getKey()); + doLoop(); } finally { isInEventLoop = false; } } - private void doLoop(K key) { + private void doLoop() { do { @@ -135,13 +144,9 @@ class StreamPollTask> implements Task { // allow interruption Thread.sleep(0); - List read = readFunction.apply(key, pollState.getCurrentReadOffset()); + List raw = readRecords(); + deserializeAndEmitRecords(raw); - for (V message : read) { - - listener.onMessage(message); - pollState.updateReadOffset(message.getId().getValue()); - } } catch (InterruptedException e) { cancel(); @@ -157,6 +162,43 @@ class StreamPollTask> implements Task { } while (pollState.isSubscriptionActive()); } + private List readRecords() { + return readFunction.apply(pollState.getCurrentReadOffset()); + } + + private void deserializeAndEmitRecords(List records) { + + for (ByteRecord raw : records) { + + try { + + pollState.updateReadOffset(raw.getId().getValue()); + V record = convertRecord(raw); + listener.onMessage(record); + } catch (RuntimeException e) { + + if (cancelSubscriptionOnError.test(e)) { + + cancel(); + errorHandler.handleError(e); + + return; + } + + errorHandler.handleError(e); + } + } + } + + private V convertRecord(ByteRecord record) { + + try { + return deserializer.apply(record); + } catch (RuntimeException e) { + throw new ConversionFailedException(TypeDescriptor.forObject(record), targetType, record, e); + } + } + @Override public boolean isActive() { return State.RUNNING.equals(getState()) || isInEventLoop; diff --git a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java index 771d86237..2e5f4e28c 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java @@ -16,10 +16,14 @@ package org.springframework.data.redis.stream; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import java.nio.ByteBuffer; import java.time.Duration; import java.util.OptionalInt; +import java.util.function.Function; + +import org.reactivestreams.Publisher; import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; import org.springframework.data.redis.connection.stream.Consumer; @@ -74,8 +78,12 @@ import org.springframework.util.Assert; * ({@code $}) for subsequent reads. * * Note: Using {@link ReadOffset#latest()} bears the chance of dropped records as records can arrive in the time - * during polling is suspended. Use recorddId's as offset or {@link ReadOffset#lastConsumed()} to minimize the chance of + * during polling is suspended. Use recordId's as offset or {@link ReadOffset#lastConsumed()} to minimize the chance of * record loss. + *

+ * {@link StreamReceiver} propagates errors during stream reads and deserialization as terminal error signal by default. + * Configuring a {@link StreamReceiverOptions#getResumeFunction() resume function} allows conditional resumption by + * dropping the record or by propagating the error to terminate the subscription. *

* See the following example code how to use {@link StreamReceiver}: * @@ -189,6 +197,7 @@ public interface StreamReceiver> { private final Duration pollTimeout; private final @Nullable Integer batchSize; + private final Function> resumeFunction; private final SerializationPair keySerializer; private final SerializationPair hashKeySerializer; private final SerializationPair hashValueSerializer; @@ -196,12 +205,14 @@ public interface StreamReceiver> { private final @Nullable HashMapper hashMapper; @SuppressWarnings({ "unchecked", "rawtypes" }) - private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, SerializationPair keySerializer, + private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, + Function> resumeFunction, SerializationPair keySerializer, SerializationPair hashKeySerializer, SerializationPair hashValueSerializer, @Nullable Class targetType, @Nullable HashMapper hashMapper) { this.pollTimeout = pollTimeout; this.batchSize = batchSize; + this.resumeFunction = resumeFunction; this.keySerializer = keySerializer; this.hashKeySerializer = hashKeySerializer; this.hashValueSerializer = hashValueSerializer; @@ -249,6 +260,10 @@ public interface StreamReceiver> { return batchSize != null ? OptionalInt.of(batchSize) : OptionalInt.empty(); } + public Function> getResumeFunction() { + return resumeFunction; + } + public SerializationPair getKeySerializer() { return keySerializer; } @@ -266,6 +281,19 @@ public interface StreamReceiver> { return hashMapper; } + public HashMapper getRequiredHashMapper() { + + if (!hasHashMapper()) { + throw new IllegalStateException("No HashMapper configured"); + } + + return hashMapper; + } + + public boolean hasHashMapper() { + return this.hashMapper != null; + } + public Class getTargetType() { if (this.targetType != null) { @@ -274,6 +302,7 @@ public interface StreamReceiver> { return Object.class; } + } /** @@ -288,6 +317,7 @@ public interface StreamReceiver> { private SerializationPair keySerializer; private SerializationPair hashKeySerializer; private SerializationPair hashValueSerializer; + private Function> resumeFunction = Mono::error; private @Nullable HashMapper hashMapper; private @Nullable Class targetType; @@ -311,7 +341,7 @@ public interface StreamReceiver> { /** * Configure a batch size for the {@code COUNT} option during reading. * - * @param recordsPerPoll must not be greater zero. + * @param recordsPerPoll must be greater zero. * @return {@code this} {@link StreamReceiverOptionsBuilder}. */ public StreamReceiverOptionsBuilder batchSize(int recordsPerPoll) { @@ -322,6 +352,25 @@ public interface StreamReceiver> { return this; } + /** + * Configure a resume {@link Function} to resume the main sequence when polling the stream fails. The function can + * either resume by suppressing the error or fail the main sequence by emitting the error to stop receiving. Receive + * errors (Redis errors, Serialization failures) stop receiving by default. + * + * @param resumeFunction must not be {@literal null}. + * @return {@code this} {@link StreamReceiverOptionsBuilder}. + * @since 2.x + * @see Flux#onErrorResume(Function) + */ + public StreamReceiverOptionsBuilder onErrorResume( + Function> resumeFunction) { + + Assert.notNull(resumeFunction, "Resume function must not be null"); + + this.resumeFunction = resumeFunction; + return this; + } + /** * Configure a key, hash key and hash value serializer. * @@ -450,7 +499,8 @@ public interface StreamReceiver> { * @return new {@link StreamReceiverOptions}. */ public StreamReceiverOptions build() { - return new StreamReceiverOptions<>(pollTimeout, batchSize, keySerializer, hashKeySerializer, hashValueSerializer, + return new StreamReceiverOptions<>(pollTimeout, batchSize, resumeFunction, keySerializer, hashKeySerializer, + hashValueSerializer, targetType, hashMapper); } } 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 46dc908c4..5db2b6a57 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java @@ -35,12 +35,14 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.core.convert.ConversionFailedException; import org.springframework.data.redis.SettingsUtils; 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.LettuceConnection; import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension; +import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.ObjectRecord; @@ -276,7 +278,7 @@ public class StreamMessageListenerContainerIntegrationTests { StreamReadRequest readRequest = StreamReadRequest .builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())) // - .errorHandler(failures::add) // // + .errorHandler(failures::add) // .cancelOnError(t -> false) // .consumer(Consumer.from("my-group", "my-consumer")) // .build(); @@ -298,6 +300,52 @@ public class StreamMessageListenerContainerIntegrationTests { cancelAwait(subscription); } + @Test // DATAREDIS-1230 + void deserializationShouldContinueStreamRead() throws InterruptedException { + + StreamMessageListenerContainerOptions> containerOptions = StreamMessageListenerContainerOptions + .builder().batchSize(1).pollTimeout(Duration.ofMillis(100)).targetType(Long.class).build(); + + BlockingQueue> records = new LinkedBlockingQueue<>(); + BlockingQueue failures = new LinkedBlockingQueue<>(); + + StreamMessageListenerContainer> container = StreamMessageListenerContainer + .create(connectionFactory, containerOptions); + + StreamReadRequest readRequest = StreamReadRequest + .builder(StreamOffset.create("my-stream", ReadOffset.from("0-0"))) // + .errorHandler(failures::add) // + .cancelOnError(t -> false) // + .build(); + + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "1")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "foo")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "3")); + + container.start(); + Subscription subscription = container.register(readRequest, records::add); + + subscription.await(DEFAULT_TIMEOUT); + + ObjectRecord first = records.poll(1, TimeUnit.SECONDS); + Throwable conversionFailure = failures.poll(1, TimeUnit.SECONDS); + ObjectRecord third = records.poll(1, TimeUnit.SECONDS); + + assertThat(first).isNotNull(); + assertThat(first.getValue()).isEqualTo(1L); + + assertThat(conversionFailure).isInstanceOf(ConversionFailedException.class) + .hasCauseInstanceOf(ConversionFailedException.class).hasRootCauseInstanceOf(NumberFormatException.class); + assertThat(((ConversionFailedException) conversionFailure).getValue()).isInstanceOf(ByteRecord.class); + + assertThat(third).isNotNull(); + assertThat(third.getValue()).isEqualTo(3L); + + assertThat(subscription.isActive()).isTrue(); + + cancelAwait(subscription); + } + @Test // DATAREDIS-864 void cancelledStreamShouldNotReceiveMessages() throws InterruptedException { diff --git a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java index 3fc83413f..60ed981df 100644 --- a/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java @@ -22,24 +22,29 @@ import static org.mockito.Mockito.*; import lombok.AllArgsConstructor; import lombok.Data; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import java.nio.ByteBuffer; import java.time.Duration; import java.util.Collections; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.core.convert.ConversionFailedException; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension; +import org.springframework.data.redis.connection.stream.ByteBufferRecord; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.ObjectRecord; import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.Record; import org.springframework.data.redis.connection.stream.StreamOffset; import org.springframework.data.redis.core.ReactiveRedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; @@ -254,6 +259,37 @@ public class StreamReceiverIntegrationTests { .verify(Duration.ofSeconds(5)); } + @Test // DATAREDIS-864 + void shouldResumeFromError() { + + AtomicReference ref = new AtomicReference<>(); + StreamReceiverOptions> options = StreamReceiverOptions.builder() + .pollTimeout(Duration.ofMillis(100)).targetType(Long.class).onErrorResume(throwable -> { + + ref.set(throwable); + return Mono.empty(); + }).build(); + + StreamReceiver> receiver = StreamReceiver.create(connectionFactory, options); + + Flux> messages = receiver.receive(StreamOffset.fromStart("my-stream")); + + redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from("0-0"), "my-group"); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "1")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "foo")); + redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("payload", "3")); + + messages.map(Record::getValue).as(StepVerifier::create) // + .expectNext(1L) // + .expectNext(3L) // + .thenCancel() // + .verify(); + + assertThat(ref.get()).isInstanceOf(ConversionFailedException.class) + .hasCauseInstanceOf(ConversionFailedException.class).hasRootCauseInstanceOf(NumberFormatException.class); + assertThat(((ConversionFailedException) ref.get()).getValue()).isInstanceOf(ByteBufferRecord.class); + } + @Data @AllArgsConstructor static class LoginEvent {