From d9340030bbdb5ca6e17643ce48c03927aaf88c1a Mon Sep 17 00:00:00 2001 From: Vedran Pavic Date: Thu, 23 Mar 2017 00:18:33 +0100 Subject: [PATCH] INT-4246: Revoke leader when lock isn't acquired JIRA: https://jira.spring.io/browse/INT-4246 The `LockRegistryLeaderInitiator` currently does not revoke leadership when leading instance is unable to acquire lock from the underlying lock. This can result in multiple `LockRegistryLeaderInitiator` instances becoming leaders in the exceptional situations such as lock timeouts. * Handle leadership revoking when leading `LockRegistryLeaderInitiator` is unable to acquire lock Polish javadoc **Cherry-pick to master** --- .../leader/LockRegistryLeaderInitiator.java | 43 +++++++----- .../LockRegistryLeaderInitiatorTests.java | 68 ++++++++++++++++++- .../jdbc/lock/JdbcLockRegistry.java | 2 +- 3 files changed, 93 insertions(+), 20 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 5433ba0f57..86d28768d8 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 the original author or authors. + * Copyright 2016-2017 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. @@ -55,6 +55,7 @@ import org.springframework.util.Assert; * * @author Dave Syer * @author Artem Bilan + * @author Vedran Pavic * @since 4.3.1 */ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware { @@ -323,11 +324,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe if (acquired) { // Success: we are now leader this.locked = true; - LockRegistryLeaderInitiator.this.candidate.onGranted(this.context); - if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) { - LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnGranted( - LockRegistryLeaderInitiator.this, this.context, this.lockKey); - } + handleGranted(); } } else if (acquired) { @@ -338,6 +335,9 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe 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); } @@ -347,12 +347,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe this.lock.unlock(); this.locked = false; // The lock was broken and we are no longer leader - LockRegistryLeaderInitiator.this.candidate.onRevoked(this.context); - if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) { - LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnRevoked( - LockRegistryLeaderInitiator.this, this.context, - LockRegistryLeaderInitiator.this.candidate.getRole()); - } + handleRevoked(); // Give it a chance to elect some other leader. Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); Thread.currentThread().interrupt(); @@ -365,12 +360,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe this.lock.unlock(); if (this.locked) { // We are stopping, therefore not leading any more - LockRegistryLeaderInitiator.this.candidate.onRevoked(this.context); - if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) { - LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnRevoked( - LockRegistryLeaderInitiator.this, this.context, - LockRegistryLeaderInitiator.this.candidate.getRole()); - } + handleRevoked(); } this.locked = false; } @@ -381,6 +371,23 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe return this.locked; } + private void handleGranted() throws InterruptedException { + LockRegistryLeaderInitiator.this.candidate.onGranted(this.context); + if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) { + LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnGranted( + LockRegistryLeaderInitiator.this, this.context, this.lockKey); + } + } + + private void handleRevoked() { + LockRegistryLeaderInitiator.this.candidate.onRevoked(this.context); + if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) { + LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnRevoked( + LockRegistryLeaderInitiator.this, this.context, + LockRegistryLeaderInitiator.this.candidate.getRole()); + } + } + } /** 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 f971252552..00d7ac0266 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-2016 the original author or authors. + * Copyright 2012-2017 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. @@ -18,9 +18,16 @@ package org.springframework.integration.support.leader; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyLong; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.mock; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; import org.apache.log4j.Level; import org.junit.Before; @@ -37,6 +44,7 @@ import org.springframework.integration.test.rule.Log4jLevelAdjuster; /** * @author Dave Syer * @author Artem Bilan + * @author Vedran Pavic * * @since 4.3.1 */ @@ -100,6 +108,64 @@ public class LockRegistryLeaderInitiatorTests { assertThat(another.getContext().isLeader(), is(true)); } + @Test + public void competingWithLock() throws Exception { + // switch used to toggle which registry obtains lock + AtomicBoolean firstLocked = new AtomicBoolean(true); + + // set up first registry instance - this one will be able to obtain lock initially + LockRegistry firstRegistry = mock(LockRegistry.class); + Lock firstLock = mock(Lock.class); + given(firstRegistry.obtain(anyString())).willReturn(firstLock); + given(firstLock.tryLock(anyLong(), any(TimeUnit.class))).willAnswer(i -> firstLocked.get()); + + // set up first initiator instance using first LockRegistry + LockRegistryLeaderInitiator first = + new LockRegistryLeaderInitiator(firstRegistry, new DefaultCandidate()); + CountDownLatch firstGranted = new CountDownLatch(1); + CountDownLatch firstRevoked = new CountDownLatch(1); + first.setHeartBeatMillis(10); + first.setBusyWaitMillis(1); + first.setLeaderEventPublisher(new CountingPublisher(firstGranted, firstRevoked)); + + // set up second registry instance - this one will NOT be able to obtain lock initially + LockRegistry secondRegistry = mock(LockRegistry.class); + Lock secondLock = mock(Lock.class); + given(secondRegistry.obtain(anyString())).willReturn(secondLock); + given(secondLock.tryLock(anyLong(), any(TimeUnit.class))).willAnswer(i -> !firstLocked.get()); + + // set up second initiator instance using second LockRegistry + LockRegistryLeaderInitiator second = + new LockRegistryLeaderInitiator(secondRegistry, new DefaultCandidate()); + CountDownLatch secondGranted = new CountDownLatch(1); + CountDownLatch secondRevoked = new CountDownLatch(1); + second.setHeartBeatMillis(10); + second.setBusyWaitMillis(1); + second.setLeaderEventPublisher(new CountingPublisher(secondGranted, secondRevoked)); + + // start initiators + first.start(); + second.start(); + + Thread.sleep(100); + + // first initiator should lead and publish granted event + assertThat(first.getContext().isLeader(), is(true)); + assertThat(second.getContext().isLeader(), is(false)); + assertThat(firstGranted.await(10, TimeUnit.SECONDS), is(true)); + + // simulate first registry instance unable to obtain lock, for example due to lock timeout + firstLocked.set(false); + + Thread.sleep(100); + + // second initiator should take lead and publish granted event, first initiator should publish revoked event + assertThat(second.getContext().isLeader(), is(true)); + assertThat(first.getContext().isLeader(), is(false)); + assertThat(secondGranted.await(10, TimeUnit.SECONDS), is(true)); + assertThat(firstRevoked.await(10, TimeUnit.SECONDS), is(true)); + } + private static class CountingPublisher implements LeaderEventPublisher { private final CountDownLatch granted; diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java index 8404a710ab..df7ec9cd31 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java @@ -44,7 +44,7 @@ import org.springframework.util.Assert; * * @author Dave Syer * @author Artem Bilan - * @author Vedran Pavić + * @author Vedran Pavic * * @since 4.3 */