Fix redundant disposal in LettucePoolingConnectionProvider by implementing SmartLifecycle #3100

[mayleaf]

Enhancements to LettuceConnectionFactory:

• Added proper disposal of ClusterCommandExecutor during the stop method to ensure resources are released cleanly.
• Removed redundant cleanup logic for ClusterCommandExecutor from the destroy method, as it is now handled in the stop method.
• Introduced a new private dispose method to encapsulate the cleanup logic for ClusterCommandExecutor, improving code readability and maintainability.

Enhancements to LettucePoolingConnectionProvider:

• Implemented SmartLifecycle to manage the lifecycle of the connection provider, including start, stop, and isRunning methods.
• Added an AtomicReference-based State enum to track the lifecycle state of the connection provider, ensuring thread-safe transitions between states.
  These updates improve the robustness and maintainability of the connection handling logic in the Spring Data Redis project.

---

I have found out that #3100 was reproduced when redis run with cluster mode.

LettuceConnectionFactory uses a connectionProvider that refered at ClusterCommandExecuter. When redis run with Cluster mode, it uses LettucePoolingConnectionProvider, which only implements DisposableBean.

When the LettuceConnectionFactory disposed, it disposes ConnectionProvider first, and disposes ClusterCommandExecuter next.

And when the ClusterCommandExecuter disposed, it disposes the connectionProvider - shared with LettuceConnectionFactory. As a result, LettucePoolingConnectionProvider's destroy method called twice.

Finally, the LettucePoolingConnectionProvider tried releasing connections in it's pool - that has 0 connections.

To solve this issue, I've implemented SmartLifeCycle at LettucePoolingConnectionProvider, and prevent duplicated stop method execution by AtomicReference based state management.
Solved #3100, See #3103

• You have read the Spring Data contribution guidelines.
• You use the code formatters provided here and have them applied to your changes. Don’t submit any formatting related changes.
• You submit test cases (unit or integration tests) that back your changes.
• You added yourself as author in the headers of the classes you touched. Amend the date range in the Apache license header if needed. For new types, add the license header (copy from another file and set the current year only).

[mp911de]

In these changes, ClusterCommandExecutor is being disposed after disposing the connection provider and so there's no behavioral change. Shouldn't clusterCommandExecutor be disposed (and thus its shared connection returned to the pool) before disposing connectionProvider?

[mayleaf]

You’re right that there’s no behavioral change in ClusterCommandExecutor itself. However, despite other resources at LettuceConnectionFactory are disposed at stop, only ClusterCommandExecutor disposed at destroy method. I was concerned that if we leave the dispose call in destroy, it could be executed twice in the future, as LettucePoolingConnectionProvider did. Since the stop method uses an AtomicReference guard, this change ensures safe and idempotent disposal.

[mayleaf]

And I believe the order of disposing connectionProvider doesn’t affect behavior in this case.

Below are the previous implementations of LettucePoolingConnectionProvider#destroy, #release As you can see, this method fully clears all pools and references on the first invocation. If it were called again, it could potentially throw exceptions or behave unexpectedly due to already-released resources.

    @Override
	public void destroy() throws Exception {

		List<CompletableFuture<?>> futures = new ArrayList<>();
		if (!poolRef.isEmpty() || !asyncPoolRef.isEmpty()) {
			log.warn("LettucePoolingConnectionProvider contains unreleased connections");
		}

		if (!inProgressAsyncPoolRef.isEmpty()) {

			log.warn("LettucePoolingConnectionProvider has active connection retrievals");
			inProgressAsyncPoolRef.forEach((k, v) -> futures.add(k.thenApply(StatefulConnection::closeAsync)));
		}

		if (!poolRef.isEmpty()) {

			poolRef.forEach((connection, pool) -> pool.returnObject(connection));
			poolRef.clear();
		}

		if (!asyncPoolRef.isEmpty()) {

			asyncPoolRef.forEach((connection, pool) -> futures.add(pool.release(connection)));
			asyncPoolRef.clear();
		}

		pools.forEach((type, pool) -> pool.close());

		CompletableFuture
				.allOf(futures.stream().map(it -> it.exceptionally(LettuceFutureUtils.ignoreErrors()))
						.toArray(CompletableFuture[]::new)) //
				.thenCompose(ignored -> {

					CompletableFuture[] poolClose = asyncPools.values().stream().map(AsyncPool::closeAsync)
							.map(it -> it.exceptionally(LettuceFutureUtils.ignoreErrors())).toArray(CompletableFuture[]::new);

					return CompletableFuture.allOf(poolClose);
				}) //
				.thenRun(() -> {
					asyncPoolRef.clear();
					inProgressAsyncPoolRef.clear();
				}) //
				.join();

		pools.clear();
	}    

    @Override
	public void release(StatefulConnection<?, ?> connection) {

		GenericObjectPool<StatefulConnection<?, ?>> pool = poolRef.remove(connection);

		if (pool == null) {

			AsyncPool<StatefulConnection<?, ?>> asyncPool = asyncPoolRef.remove(connection);

			if (asyncPool == null) {
				throw new PoolException("Returned connection " + connection
						+ " was either previously returned or does not belong to this connection provider");
			}

			discardIfNecessary(connection);
			asyncPool.release(connection).join();
			return;
		}

		discardIfNecessary(connection);
		pool.returnObject(connection);
	}

To prevent this, the disposal logic was moved to under the state safe guard, ensuring it’s executed only once and making the lifecycle handling more robust.