Nullability of RedisElementReader.read(…) contradicts non-nullability of Flux.map(…)

[zhou-hao]

RedisElementReader.read allow return null,but ReactiveXXXOperations using Flux.map will result in The mapper [......] returned a null value. errors.

spring-data-redis/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java

Lines 339 to 345 in 5568ffd

	public Flux<V> members(K key) {
	Assert.notNull(key, "Key must not be null");
	return createFlux(connection -> connection.sMembers(rawKey(key)).map(this::readValue));
	}

In my case, RedisElementReader.read well return null when illegal data format.

Is using .mapNotNull better ?

[mp911de]

If you would like us to spend some time helping you to diagnose the problem, please provide a minimal sample that reproduces the problem.

[zhou-hao]

I create a repository for reproduce this issue.

https://github.com/zhou-hao/spring-data-redis-issue2655/blob/d40dc0bab18a6f0d5215ba8b1aef5a8ec4bbd01d/src/test/java/com/example/demo/DemoApplicationTests.java#L24-L76

 @Test
    void testRedisElementReaderReturnNull() {

        LettuceConnectionFactory factory
                = new LettuceConnectionFactory(
                LettuceConnectionFactory
                        .createRedisConfiguration("redis://localhost:6379")
        );
        factory.afterPropertiesSet();
        ObjectRedisElementReader reader = new ObjectRedisElementReader();

        RedisElementReader<String> keyReader = RedisElementReader.from(RedisSerializer.string());
        RedisElementWriter<String> keyWriter = RedisElementWriter.from(RedisSerializer.string());

        RedisSerializationContext<String, Object> serializationContext = RedisSerializationContext
                .<String, Object>newSerializationContext()
                .key(keyReader, keyWriter)
                .value(reader, RedisElementWriter.from(RedisSerializer.java()))
                .hashKey(keyReader, keyWriter)
                .hashValue(reader, RedisElementWriter.from(RedisSerializer.java()))
                .build();
        ReactiveRedisTemplate<String, Object> template = new ReactiveRedisTemplate<>(factory, serializationContext);

        template.opsForSet()
                .add("test-set", "a", "b", "c")
                .block();

        List<Object> test = template
                .opsForSet()
                .members("test-set")
                .collectList()
                .block();

        Assertions.assertNotNull(test);
        System.out.println(test);
        Assertions.assertEquals(2, test.size());
    }


    static class ObjectRedisElementReader implements RedisElementReader<Object> {

        static final RedisElementReader<Object> reader = RedisElementReader.from(RedisSerializer.java());

        @Override
        @Nullable
        public Object read(ByteBuffer buffer) {
            Object val = reader.read(buffer);
            // mock return null value
            return Objects.equals("b", val) ? null : val;
        }

    }

[mp911de]

Thanks for your example. In some methods, returning null from RedisElementReader.read(…) is fine as these elements are collected in a collection that permits null values (e.g., mget).

Reactive Streams prohibit the use of null values, and we cannot simply change method signatures.

Filtering invalid elements from a result sounds like an application logic rule, not an infrastructure concern. Therefore, I suggest filtering the stream using the filter(…) operator.

If we filter elements on the stream decoder, the returned stream would no longer represent the underlying Redis data. For a clean implementation of both API contracts, it seems best to use mapNotNull(…) as you suggested and to document the effect of RedisElementReader<T> returning null values on streams returning Flux<T>.

[jxblum]

@zhou-hao - The key to solving your problem involves the registration of a proper RedisSerializer to handle values coming from a Redis list or set that may be null.

NOTE: Alternatively, you can register a custom SD Redis RedisElementReader.

It is entirely possible and valid for a Redis list or set to contain null elements. I demonstrated this by creating a Redis list containing null elements in my test class.

As @mp911de communicated above, the Reactive Streams specification (Rule 2.13; scroll down) prevents the emission of null values (signals) in a stream. Specifically, "The reactive streams specification disallows null values in a sequence.", which is well documented online and in frameworks, like Project Reactor (docs), upon which Spring Data Redis (and Lettuce) build on for reactive support.

Even when using the Flux.filter(:Predicate<T>) operator or the Flux.handle(:BiConsumer<T, SynchronousdSink>) operator will not work when the stream data source contains nulls.

There is a nice blob post (specifically on NullPointerExceptions) posted by baeldung on this very subject.

The only way to properly address this problem is with a RedisSerializer that handles possible null values (in the case of a Mono) or null elements coming from Redis lists/sets (in the case of a Flux) BEFORE the Reactive types ever see null values/elements in the stream. See here; line 239 is the solution!

Of course, you should implement custom null handling specific to your application, and appropriately by element/value type.

To make this a bit more concrete, this test case demonstrates the problem and why Spring Data Redis cannot do anything about it:

@Test
void fluxWithNullStreamOfElements() {

    AtomicReference<Optional<RuntimeException>> causeReference = 
        new AtomicReference<>(Optional.empty());

    Flux.fromIterable(Arrays.asList(1, 2, null, 4, null, null, null, 8, null))
        .filter(Objects::nonNull)
        .handle((value, sink) -> sink.next(value != null ? value : 0))
        .subscribe(System.out::println, cause -> causeReference.set(
            Optional.of(new RuntimeException("Reactive Stream processing failed", cause))
        ));

    causeReference.get().ifPresent(cause -> { throw cause; });
}

This leads to the following output and Exception:

0
1
2

java.lang.RuntimeException: Reactive Stream processing failed
  at io.vmware.spring.data.redis.tests.template.RedisTemplateIntegrationTests
    .lambda$fluxWithNullStreamOfElements$3(RedisTemplateIntegrationTests.java:217)
  at reactor.core.publisher.LambdaSubscriber.onError(LambdaSubscriber.java:149)
  at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber
    .onError(FluxContextWrite.java:121)
  at reactor.core.publisher.FluxHandleFuseable$HandleFuseableConditionalSubscriber
    .onError(FluxHandleFuseable.java:595)
  at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber
    .onError(FluxFilterFuseable.java:382)
  at reactor.core.publisher.FluxIterable$IterableSubscriptionConditional
    .fastPath(FluxIterable.java:724)
  at reactor.core.publisher.FluxIterable$IterableSubscriptionConditional
    .request(FluxIterable.java:620)
  at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber
    .request(FluxFilterFuseable.java:411)
  at reactor.core.publisher.FluxHandleFuseable$HandleFuseableConditionalSubscriber
    .request(FluxHandleFuseable.java:654)
  at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber
    .request(FluxContextWrite.java:136)
  at reactor.core.publisher.LambdaSubscriber.onSubscribe(LambdaSubscriber.java:119)
  at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber
    .onSubscribe(FluxContextWrite.java:101)
  at reactor.core.publisher.FluxHandleFuseable$HandleFuseableConditionalSubscriber
    .onSubscribe(FluxHandleFuseable.java:473)
  at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber
    .onSubscribe(FluxFilterFuseable.java:305)
  at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:179)
  at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:83)
  at reactor.core.publisher.Flux.subscribe(Flux.java:8773)
  at reactor.core.publisher.Flux.subscribeWith(Flux.java:8894)
  at reactor.core.publisher.Flux.subscribe(Flux.java:8739)
  at reactor.core.publisher.Flux.subscribe(Flux.java:8663)
  at reactor.core.publisher.Flux.subscribe(Flux.java:8633)
  at io.vmware.spring.data.redis.tests.template.RedisTemplateIntegrationTests
    .fluxWithNullStreamOfElements(RedisTemplateIntegrationTests.java:216)
  at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
...
Caused by: java.lang.NullPointerException: The iterator returned a null value
  at java.base/java.util.Objects.requireNonNull(Objects.java:233)
  at reactor.core.publisher.FluxIterable$IterableSubscriptionConditional
    .fastPath(FluxIterable.java:720)
	... 86 more

Therefore, you MUST implement a custom RedisSerializer and handle the data properly yourself when you are using any Reactive library/framework and your backend Redis database stores null values!

[mp911de]

I reinstated the bug label as status: invalid means that we do not work on a ticket because it isn't valid.