NullPointerException thrown during AbstractRedisAsyncCommands.flushCommands

[dev-konrad]

Bug Report

Possibly there is a bug in lettuce, where NPE is thrown during flushCommands operation. See some simplified code, where this is happening. I have included all custom options we set on the client. This code runs against clustered enterprise redis.
I understand that it might be also caused by our environment or some specific options we use, but we're blocked in investigation with hardly any ideas on how to proceed with it. Code runs in grid and issue appears randomly without ways to debug it. Therefore if you don't have any suggestions on what to change, could you please at least share ideas on how to improve the situation by let's say extra logging or similar?

Current Behavior

The application throws with NullPointerException, when executing AbstractRedisAsyncCommands.flushCommands

Stack trace

	at x.x.x.x.x.X.executeAll(PipelineExecutor.java:97)
	... // some more calls done in application
Caused by: java.lang.NullPointerException
	at io.lettuce.core.internal.AsyncConnectionProvider$Sync.doWithConnection(AsyncConnectionProvider.java:266)
	at io.lettuce.core.internal.AsyncConnectionProvider.lambda$forEach$3(AsyncConnectionProvider.java:190)
	at java.util.concurrent.ConcurrentHashMap$ValuesView.forEach(ConcurrentHashMap.java:4707)
	at io.lettuce.core.internal.AsyncConnectionProvider.forEach(AsyncConnectionProvider.java:190)
	at io.lettuce.core.cluster.PooledClusterConnectionProvider.flushCommands(PooledClusterConnectionProvider.java:555)
	at io.lettuce.core.cluster.ClusterDistributionChannelWriter.flushCommands(ClusterDistributionChannelWriter.java:377)
	at io.lettuce.core.RedisChannelHandler.flushCommands(RedisChannelHandler.java:326)
	at io.lettuce.core.AbstractRedisAsyncCommands.flushCommands(AbstractRedisAsyncCommands.java:539)
	at x.x.x.x.x.X.executeAll(PipelineExecutor.java:55)
	... 19 more

Input Code

Input Code

public class PipelineExecutorClass<T> {

    public Map<String, T> execute() {
		
        ClusterClientOptions.Builder optionsBuilder = ClusterClientOptions.builder();
        optionsBuilder.autoReconnect(true).cancelCommandsOnReconnectFailure(true);
        optionsBuilder.disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
                    .topologyRefreshOptions(ClusterTopologyRefreshOptions.builder().enablePeriodicRefresh(Duration.ofSeconds(5))
                            .enableAllAdaptiveRefreshTriggers().build());
        // set SSL options
        ClusterClientOptions options = optionsBuilder.build()
        
        RedisURI redisUri = RedisURI.create("redis_url");
        // set password
        io.lettuce.core.cluster.RedisClusterClient client = io.lettuce.core.cluster.RedisClusterClient.create(uri);
        client.setOptions(options);
		
        StatefulConnection<String, String> connection = client.connect.get();
        AbstractRedisAsyncCommands<String, T> advancedClusterAsyncCommands = ((StatefulRedisClusterConnection<String, String>) connection).async()
        advancedClusterAsyncCommands.setAutoFlushCommands(false);

        int attempts = 0;
        Set<String> keySet = Sets.newHashSet(/* some keys */);
		
        Map<String, RedisFuture<T>> futures = Maps.newHashMap();
        for (String key : keySet) {
            try {
                futures.put(key, advancedClusterAsyncCommands.get(key));
            } catch (Exception e) {
                // Exception occurred while setting key 
            }
        }
		Map<String, T> result = Maps.newHashMap();
        try {
            do {
                advancedClusterAsyncCommands.flushCommands();
                boolean ready = LettuceFutures.awaitAll(30000, TimeUnit.MILLISECONDS,
                        futures.values().toArray(new RedisFuture[futures.size()]));

                for (Map.Entry<String, RedisFuture<T>> item : futures.entrySet()) {
                    final RedisFuture<T> future = item.getValue();

                    if (future.isDone()) {
                        // add to result and handle errors
                    }
                }

                if (!ready) {
                    // "Timeout. Will retry to handle following keys again: ...
                }
                attempts++;
            } while (!futures.isEmpty() && attempts < 3);
        } catch (Exception e) {
            // Exception occurred while handling keys
        } finally {
            advancedClusterAsyncCommands.setAutoFlushCommands(true);
        }
        return result;
    }
}

Expected behavior/code

Application should not throw NPE.

Environment

• Lettuce version(s): [e.g. 5.2.1.RELEASE]
• Redis enterprise version: [e.g. 5.0.5], clustered

Possible Solution

Additional context

[mp911de]

Thanks for the report. Looking at the code, there should be no way that the action callback object is null. We're going to add a guard to the execution callback. Other than poking at the code, there's nothing really what you could do.

[dev-konrad]

Thanks for answering Mark. Feel free to close this if you don't have any other suggestions.
The only other option I'm left with are retries I guess. I have to decide on which level to retry though. My initial thought is that whole process must be restarted as the connection might be somehow malformed at this time and next attempt to get data will fail too.

BR,
Konrad

[mp911de]

If this issue comes back, then I'd advise you to introduce a few log/debug lines in a custom build of Lettuce to trace where the issue comes from. Closing the issue for now as we cannot do here without further knowledge.

[dev-konrad]

Thanks, and speaking of some logs - do you have any suggestions where can I add them to get some information that could help tracking the bug?