From 3d9e3c3860e1980edc50b5957c181f76e79a61b2 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 14 Dec 2023 14:00:11 +0100 Subject: [PATCH] Differentiate between initial exception handling, recovery and recovery after subscription. We now differentiate exception handling regarding the recovery state. Initial listen fails if the connection is unavailable. Upon recovery after a preceeding subscription we now log the success to create a counterpart to our error logging. Closes: #2782 Original Pull Request: #2808 # Conflicts: # src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java # src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java --- .../RedisMessageListenerContainer.java | 83 ++++++++++++++++--- ...edisMessageListenerContainerUnitTests.java | 69 +++++++++++++++ 2 files changed, 140 insertions(+), 12 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java index f26d06b48..7c0583329 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -371,7 +371,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab State state = this.state.get(); CompletableFuture futureToAwait = state.isPrepareListening() ? containerListenFuture - : lazyListen(this.backOff.start()); + : lazyListen(new InitialBackoffExecution(this.backOff.start())); try { futureToAwait.get(getMaxSubscriptionRegistrationWaitingTime(), TimeUnit.MILLISECONDS); @@ -537,8 +537,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab future.get(getMaxSubscriptionRegistrationWaitingTime(), TimeUnit.MILLISECONDS); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); - } catch (ExecutionException | TimeoutException ignore) { - } + } catch (ExecutionException | TimeoutException ignore) {} } @Override @@ -890,8 +889,13 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab Runnable recoveryFunction = () -> { - CompletableFuture lazyListen = lazyListen(backOffExecution); - lazyListen.whenComplete(propagate(future)); + CompletableFuture lazyListen = lazyListen(new RecoveryBackoffExecution(backOffExecution)); + lazyListen.whenComplete(propagate(future)).thenRun(() -> { + + if (backOffExecution instanceof RecoveryAfterSubscriptionBackoffExecution) { + logger.info("Subscription(s) recovered"); + } + }); }; if (potentiallyRecover(loggingBackOffExecution, recoveryFunction)) { @@ -984,7 +988,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab private Subscriber getRequiredSubscriber() { Assert.state(this.subscriber != null, - "Subscriber not created; Configure RedisConnectionFactory to create a Subscriber"); + "Subscriber not created; Configure RedisConnectionFactory to create a Subscriber. Make sure that afterPropertiesSet() has been called"); return this.subscriber; } @@ -1015,6 +1019,54 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } } + BackOffExecution nextBackoffExecution(BackOffExecution backOffExecution, boolean subscribed) { + + if (subscribed) { + return new RecoveryAfterSubscriptionBackoffExecution(backOff.start()); + } + + return backOffExecution; + } + + /** + * Marker for an initial backoff. + * + * @param delegate + */ + record InitialBackoffExecution(BackOffExecution delegate) implements BackOffExecution { + + @Override + public long nextBackOff() { + return delegate.nextBackOff(); + } + } + + /** + * Marker for a recovery after a subscription has been active previously. + * + * @param delegate + */ + record RecoveryAfterSubscriptionBackoffExecution(BackOffExecution delegate) implements BackOffExecution { + + @Override + public long nextBackOff() { + return delegate.nextBackOff(); + } + } + + /** + * Marker for a recovery execution. + * + * @param delegate + */ + record RecoveryBackoffExecution(BackOffExecution delegate) implements BackOffExecution { + + @Override + public long nextBackOff() { + return delegate.nextBackOff(); + } + } + /** * Represents an operation that accepts three input arguments {@link SubscriptionListener}, * {@code channel or pattern}, and {@code count} and returns no result. @@ -1189,10 +1241,15 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab try { eventuallyPerformSubscription(connection, backOffExecution, initFuture, patterns, channels); } catch (Throwable t) { - handleSubscriptionException(initFuture, backOffExecution, t); + handleSubscriptionException(initFuture, nextBackoffExecution(backOffExecution, connection.isSubscribed()), + t); } } catch (RuntimeException ex) { - initFuture.completeExceptionally(ex); + if (backOffExecution instanceof InitialBackoffExecution) { + initFuture.completeExceptionally(ex); + } else { + handleSubscriptionException(initFuture, backOffExecution, ex); + } } return initFuture; @@ -1205,8 +1262,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab void eventuallyPerformSubscription(RedisConnection connection, BackOffExecution backOffExecution, CompletableFuture subscriptionDone, Collection patterns, Collection channels) { - addSynchronization(new SynchronizingMessageListener.SubscriptionSynchronization(patterns, channels, - () -> subscriptionDone.complete(null))); + addSynchronization(new SynchronizingMessageListener.SubscriptionSynchronization(patterns, channels, () -> { + subscriptionDone.complete(null); + })); doSubscribe(connection, patterns, channels); } @@ -1412,7 +1470,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab try { subscribeChannel(channels.toArray(new byte[0][])); } catch (Exception ex) { - handleSubscriptionException(subscriptionDone, backOffExecution, ex); + handleSubscriptionException(subscriptionDone, nextBackoffExecution(backOffExecution, true), ex); } })); } else { @@ -1429,7 +1487,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab closeConnection(); unsubscribeFuture.complete(null); } catch (Throwable cause) { - handleSubscriptionException(subscriptionDone, backOffExecution, cause); + handleSubscriptionException(subscriptionDone, + nextBackoffExecution(backOffExecution, connection.isSubscribed()), cause); } }); } diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java index af60d8096..e9465a972 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java @@ -19,11 +19,15 @@ import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.connection.RedisConnection; @@ -33,6 +37,7 @@ import org.springframework.data.redis.connection.SubscriptionListener; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.data.redis.listener.adapter.RedisListenerExecutionFailedException; +import org.springframework.util.backoff.FixedBackOff; /** * Unit tests for {@link RedisMessageListenerContainer}. @@ -148,6 +153,70 @@ class RedisMessageListenerContainerUnitTests { assertThat(container.isListening()).isFalse(); } + @Test // GH-2335 + void shouldRecoverFromConnectionFailure() throws Exception { + + AtomicInteger requestCount = new AtomicInteger(); + AtomicBoolean shouldThrowSubscriptionException = new AtomicBoolean(); + + container = new RedisMessageListenerContainer(); + container.setConnectionFactory(connectionFactoryMock); + container.setBeanName("container"); + container.setTaskExecutor(new SyncTaskExecutor()); + container.setSubscriptionExecutor(new SimpleAsyncTaskExecutor()); + container.setMaxSubscriptionRegistrationWaitingTime(1000); + container.setRecoveryBackoff(new FixedBackOff(1, 5)); + container.afterPropertiesSet(); + + doAnswer(it -> { + + int req = requestCount.incrementAndGet(); + if (req == 1 || req == 3) { + return connectionMock; + } + + throw new RedisConnectionFailureException("Booh"); + }).when(connectionFactoryMock).getConnection(); + + CountDownLatch exceptionWait = new CountDownLatch(1); + CountDownLatch armed = new CountDownLatch(1); + CountDownLatch recoveryArmed = new CountDownLatch(1); + + doAnswer(it -> { + + SubscriptionListener listener = it.getArgument(0); + when(connectionMock.isSubscribed()).thenReturn(true); + + listener.onChannelSubscribed("a".getBytes(StandardCharsets.UTF_8), 1); + + armed.countDown(); + exceptionWait.await(); + + if (shouldThrowSubscriptionException.compareAndSet(true, false)) { + when(connectionMock.isSubscribed()).thenReturn(false); + throw new RedisConnectionFailureException("Disconnected"); + } + + recoveryArmed.countDown(); + + return null; + }).when(connectionMock).subscribe(any(), any()); + + container.start(); + container.addMessageListener(new MessageListenerAdapter(handler), new ChannelTopic("a")); + armed.await(); + + // let an exception happen + shouldThrowSubscriptionException.set(true); + exceptionWait.countDown(); + + // wait for subscription recovery + recoveryArmed.await(); + + assertThat(recoveryArmed.getCount()).isZero(); + + } + @Test // GH-964 void failsOnDuplicateInit() { assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> container.afterPropertiesSet());