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>.