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 12b1b98f90..9e4a169209 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 @@ -381,8 +381,15 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe } catch (Exception e) { if (this.locked) { - this.lock.unlock(); this.locked = false; + try { + this.lock.unlock(); + } + catch (Exception e1) { + logger.debug("Could not unlock - treat as broken. " + + "Revoking " + (isRunning() ? " and retrying..." : "..."), e); + + } // The lock was broken and we are no longer leader handleRevoked(); if (isRunning()) { @@ -408,11 +415,16 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe } finally { if (this.locked) { - this.lock.unlock(); + this.locked = false; + try { + this.lock.unlock(); + } + catch (Exception e) { + logger.debug("Could not unlock during stop - treat as broken. Revoking...", e); + } // We are stopping, therefore not leading any more handleRevoked(); } - this.locked = false; } return null; } 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 7592427854..21bdccecfa 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 @@ -167,6 +167,35 @@ public class JdbcLockRegistryLeaderInitiatorTests { initiator1.stop(); } + @Test + public void testLostConnection() throws InterruptedException { + CountDownLatch granted = new CountDownLatch(1); + CountingPublisher countingPublisher = new CountingPublisher(granted); + + DefaultLockRepository lockRepository = new DefaultLockRepository(dataSource); + lockRepository.afterPropertiesSet(); + LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(new JdbcLockRegistry(lockRepository)); + initiator.setLeaderEventPublisher(countingPublisher); + + initiator.start(); + + assertThat(granted.await(10, TimeUnit.SECONDS), is(true)); + + destroy(); + + assertThat(countingPublisher.revoked.await(10, TimeUnit.SECONDS), is(true)); + + granted = new CountDownLatch(1); + countingPublisher = new CountingPublisher(granted); + initiator.setLeaderEventPublisher(countingPublisher); + + init(); + + assertThat(granted.await(10, TimeUnit.SECONDS), is(true)); + + initiator.stop(); + } + private static class CountingPublisher implements LeaderEventPublisher { private final CountDownLatch granted;