Result is lost when published on another executor

[trueinsider]

Bug Report

Current Behavior

#905 This change introduces some kind of bug where result is lost

Input Code

RedisClient client1 = RedisClient.create("redis://localhost");
RedisClient client2 = RedisClient.create("redis://localhost");
client1.setOptions(ClientOptions.builder().publishOnScheduler(false).build());
client2.setOptions(ClientOptions.builder().publishOnScheduler(true).build());
RedisReactiveCommands<String, String> redis1 = client1.connect().reactive();
RedisReactiveCommands<String, String> redis2 = client2.connect().reactive();
int counter1 = 0;
int counter2 = 0;
for (int i = 0; i < 1000; i++) {
    if (redis1.eval("return 1", INTEGER).next().block() == null) counter1++;
    if (redis2.eval("return 1", INTEGER).next().block() == null) counter2++;
}
out.println(counter1);
out.println(counter2);

Expected behavior/code

Basically, counter1 == counter2 == 0 should be true, but counter2 has value 1-100 instead of 0.

Environment

• Lettuce version(s): 5.1.4.RELEASE
• Redis version: 5.0.0

Possible Solution

Make sure that single instance of RedisPublisher.PublishOnSubscriber will use same Executor from EventExecutorGroup for all it's operations.

Additional context

I think what happens is this:
RedisPublisher.OnNext and RedisPublisher.OnComplete are scheduled to the executor in the correct order, but since executor is non-immediate and multi-threaded race condition happens and sometimes RedisPublisher.OnComplete is executed before RedisPublisher.OnNext

[mp911de]

Awesome work, thanks a lot for looking into that. The issue supposedly happens only with Flux and not with Mono. We have already a bit of code that creates RedisPublisher:

Flux

https://github.com/lettuce-io/lettuce-core/blob/657ed8a966e0ecb305d8fd59f8b776bb3b9f9b8b/src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java#L400-L409

Mono

https://github.com/lettuce-io/lettuce-core/blob/657ed8a966e0ecb305d8fd59f8b776bb3b9f9b8b/src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java#L422-L432

The Mono code calls correctly .next() while the Flux code misses the next() call. Care to adjust your PR accordingly so we can merge it?

[trueinsider]

@mp911de sure, done

[mp911de]

Thanks a lot for all your support. The fix is scheduled for 5.1.5, expect a release by next week.

[trueinsider]

@mp911de thanks for fast merge. Can you provide snapshot for the time being?

[mp911de]

It's available for 5.2.0 and 5.1.5: https://oss.sonatype.org/content/repositories/snapshots/io/lettuce/lettuce-core/5.1.5.BUILD-SNAPSHOT/lettuce-core-5.1.5.BUILD-20190227.153141-3.jar