Initiate a single subscription on concurrent addMessageListener calls

[rlaehddus302]

Closes #3447

Problem

When RedisMessageListenerContainer is started without listeners and two threads call addMessageListener(…) at the same time, either two subscription connections are created (every message is delivered twice) or one of the channels is never subscribed. See #3447 for details and a reproducer.

Cause

1. Two connections. lazyListen() inspects the state and only afterwards calls doSubscribe(…). doSubscribe(…) reads the state again and uses whatever it just read as the expected value of compareAndSet(state, State.prepareListening()). If another thread has already moved the state to prepareListening (or listening), the CAS still succeeds and Subscriber.initialize(…) runs a second time, obtaining a second connection.
2. Skipped channel. addListener(…) captured wasListening before calling lazyListen(). A thread that waited in lazyListen() for another thread's initial subscription continued with wasListening == false and skipped subscribeChannel(…)/subscribePattern(…). If its topic was registered after the initiating thread collected the topics for the initial SUBSCRIBE, nobody subscribed to it.

Change

• doSubscribe(…) returns SubscribeResult (RETRY, INITIATED, ALREADY_ACTIVE) instead of boolean. For an initial subscription request (InitialBackoffExecution) it returns ALREADY_ACTIVE without calling Subscriber.initialize(…) if the state is already prepareListening/listening.
• Recovery requests (RecoveryBackoffExecution, RecoveryAfterSubscriptionBackoffExecution) behave as before. They have to re-subscribe while the container is in the listening state, so the new check does not apply to them.
• lazyListen(BackOffExecution) returns the future to await together with the information whether the calling thread initiated the subscription (ListenAttempt), and lazyListen() returns that flag.
• addListener(…) subscribes its topics explicitly if it did not initiate the subscription and the container is listening (!initiated && isListening()), instead of relying on the state captured before waiting.

One consequence: a caller that did not initiate the subscription may send SUBSCRIBE for a channel that the initial subscription already contains (when its topic was registered before the initiating thread collected the topics). I consider that harmless because subscribing to an already subscribed channel on the same connection does not create a second subscription, but I'd be glad to hear if you prefer a different approach.

Verification

• New unit test RedisMessageListenerContainerUnitTests.concurrentInitialAddMessageListenerShouldInitiateSingleSubscription. It holds both callers inside BackOff.start(), which lazyListen() invokes after inspecting the state and before initiating the subscription, so both callers observe "not listening" deterministically. Without the change to RedisMessageListenerContainer the test fails in 3 of 3 runs (getConnection() wanted 1 time, but was 2 times). With the change it passes (the test class was run more than ten times, all green).
• The existing tests in RedisMessageListenerContainerUnitTests pass, including shouldRecoverFromConnectionFailure.
• Reproducer from RedisMessageListenerContainer: concurrent addMessageListener calls during the initial subscription create two subscription connections or silently skip a channel #3447 (https://github.com/rlaehddus302/spring-data-redis-listener-race, Lettuce, Redis 7 standalone), run against classes built from main: 50 trials gave OK 29 / TWICE 4 / MISSING 16 before the change. With the change, three runs of 50 trials each gave TWICE 0 / MISSING 0. In both variants only the very first trial of a run failed with "Subscription registration timeout exceeded" (the container's default registration timeout of 2 seconds). I assume this is JVM/connection warm-up, but I did not investigate it.

Scope of testing

• Integration tests: I ran the tests of the org.springframework.data.redis.listener package (-Dtest=org/springframework/data/redis/listener/*Tests.java -DrunLongTests=true) against the project's Docker Compose environment (Redis 8.6.2: standalone with replicas and sentinels, auth, cluster): 115 tests, 0 failures, 0 errors, 3 skipped (the three cases that the tests themselves skip for cluster-aware connections). I did not run the rest of the test suite.
• The reproducer was only run with Lettuce against a standalone Redis. Jedis was not tested against a real server (the unit tests use a mocked connection).
• I did not test concurrent addMessageListener together with stop()/removeMessageListener (Pub/Sub in a concurrent scenario may result in the inability to use subscriptions anymore, which can ultimately lead to memory leaks #3080, Pub-Sub Race conditions on high load - Subscription has been unsubscribed and cannot be used anymore #2425); this change does not try to address those.