From 935da81ab3cc8c7a86345e683299ab02e9bf0da3 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** # Conflicts: # spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java # spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java --- .../leader/LockRegistryLeaderInitiator.java | 37 ++++++++++++------- .../LockRegistryLeaderInitiatorTests.java | 34 ++++++++++++++++- .../JdbcLockRegistryLeaderInitiatorTests.java | 6 --- 3 files changed, 56 insertions(+), 21 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 9ce0c936ed..460d8c0d1e 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. @@ -312,7 +312,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 // TODO obtain(this.lockKey) because of INT-4248. Should be fixed in 5.0 @@ -330,27 +330,41 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe // should release it LockRegistryLeaderInitiator.this.locks.obtain(this.lockKey) .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) { LockRegistryLeaderInitiator.this.locks.obtain(this.lockKey) .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; } } @@ -405,11 +419,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 1f4e263ec0..8ec01be7a0 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. @@ -214,6 +214,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 d854ad8b67..5636a94758 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 @@ -28,7 +28,6 @@ import java.util.concurrent.TimeUnit; import org.apache.log4j.Level; import org.junit.AfterClass; import org.junit.BeforeClass; -import org.junit.Rule; import org.junit.Test; import org.springframework.integration.jdbc.lock.DefaultLockRepository; @@ -37,7 +36,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.Log4jLevelAdjuster; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; @@ -52,10 +50,6 @@ public class JdbcLockRegistryLeaderInitiatorTests { public static EmbeddedDatabase dataSource; - @Rule - public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.DEBUG, "org.springframework.integration", - "org.springframework.integration.jdbc", "org.springframework.jdbc", "org.apache.derby"); - @BeforeClass public static void init() { dataSource = new EmbeddedDatabaseBuilder()