From 09aeaac4dab91df294de3017e1498f5e66c22c34 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 9 Feb 2018 13:33:52 -0500 Subject: [PATCH] INT-4396: Add retrying lock in case of exception JIRA: https://jira.spring.io/browse/INT-4396 When target distributed `Lock` implementation throws an exception, e.g. in case of no connection to the service, the `LockRegistryLeaderInitiator` exists the loop and can come back to the elections only after restart * Catch all the exception on `this.lock.tryLock()` and resubmit `LeaderSelector` for a new locking cycle if `LockRegistryLeaderInitiator.isRunning()` and `InterruptedException` * Remove diagnostics from the `JdbcLockRegistryLeaderInitiatorTests` since this fix confirms that we just didn't have a reconnect logic before when this test failed sporadically **Cherry-pick to 4.3.x** --- .../leader/LockRegistryLeaderInitiator.java | 37 ++++++++++++------- .../LockRegistryLeaderInitiatorTests.java | 34 ++++++++++++++++- .../JdbcLockRegistryLeaderInitiatorTests.java | 10 ----- 3 files changed, 56 insertions(+), 25 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java index e395797c77..c1ffd0b18d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -326,7 +326,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe @Override public Void call() throws Exception { try { - while (LockRegistryLeaderInitiator.this.running) { + while (isRunning()) { try { // We always try to acquire the lock, in case it expired boolean acquired = this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis, @@ -345,26 +345,40 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe // If we were able to acquire it but we were already locked we // should release it this.lock.unlock(); - // Give it a chance to expire. - Thread.sleep(LockRegistryLeaderInitiator.this.heartBeatMillis); + if (isRunning()) { + // Give it a chance to expire. + Thread.sleep(LockRegistryLeaderInitiator.this.heartBeatMillis); + } } else { this.locked = false; // We were not able to acquire it, therefore not leading any more handleRevoked(); - // Try again quickly in case the lock holder dropped it - Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); + if (isRunning()) { + // Try again quickly in case the lock holder dropped it + Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); + } } } - catch (InterruptedException e) { + catch (Exception e) { if (this.locked) { this.lock.unlock(); this.locked = false; // The lock was broken and we are no longer leader handleRevoked(); - // Give it a chance to elect some other leader. - Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); + if (isRunning()) { + // Give it a chance to elect some other leader. + Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); + } + } + + if (e instanceof InterruptedException) { Thread.currentThread().interrupt(); + if (isRunning()) { + logger.warn("Restarting LeaderSelector because of error.", e); + LockRegistryLeaderInitiator.this.future = + LockRegistryLeaderInitiator.this.executorService.submit(this); + } return null; } } @@ -446,11 +460,6 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe public void yield() { if (LockRegistryLeaderInitiator.this.future != null) { LockRegistryLeaderInitiator.this.future.cancel(true); - if (isRunning()) { - LockRegistryLeaderInitiator.this.future = - LockRegistryLeaderInitiator.this.executorService - .submit(LockRegistryLeaderInitiator.this.leaderSelector); - } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java index a944c466a8..f4957407b5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2017 the original author or authors. + * Copyright 2012-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -265,6 +265,38 @@ public class LockRegistryLeaderInitiatorTests { assertNull(throwable); } + @Test + public void testExceptionFromLock() throws Exception { + Lock mockLock = mock(Lock.class); + + AtomicBoolean exceptionThrown = new AtomicBoolean(); + + willAnswer(invocation -> { + if (!exceptionThrown.getAndSet(true)) { + throw new RuntimeException("lock is broken"); + } + else { + return true; + } + }).given(mockLock).tryLock(anyLong(), any(TimeUnit.class)); + + LockRegistry registry = lockKey -> mockLock; + + CountDownLatch onGranted = new CountDownLatch(1); + + LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(registry); + + initiator.setLeaderEventPublisher(new CountingPublisher(onGranted)); + + initiator.start(); + + assertTrue(onGranted.await(10, TimeUnit.SECONDS)); + assertTrue(initiator.getContext().isLeader()); + assertTrue(exceptionThrown.get()); + + initiator.stop(); + } + private static class CountingPublisher implements LeaderEventPublisher { private final CountDownLatch granted; diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java index 5ef5a5604e..7592427854 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java @@ -27,7 +27,6 @@ import java.util.concurrent.TimeUnit; import org.junit.AfterClass; import org.junit.BeforeClass; -import org.junit.Rule; import org.junit.Test; import org.springframework.integration.jdbc.lock.DefaultLockRepository; @@ -36,7 +35,6 @@ import org.springframework.integration.leader.Context; import org.springframework.integration.leader.DefaultCandidate; import org.springframework.integration.leader.event.LeaderEventPublisher; import org.springframework.integration.support.leader.LockRegistryLeaderInitiator; -import org.springframework.integration.test.rule.Log4j2LevelAdjuster; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; @@ -52,14 +50,6 @@ public class JdbcLockRegistryLeaderInitiatorTests { public static EmbeddedDatabase dataSource; - @Rule - public Log4j2LevelAdjuster adjuster = - Log4j2LevelAdjuster.trace() - .categories("org.springframework.integration", - "org.springframework.integration.jdbc", - "org.springframework.jdbc", - "org.apache.derby"); - @BeforeClass public static void init() { dataSource = new EmbeddedDatabaseBuilder()