From 51c49519dce8ffc49790181d9901a87fb13a5f8f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 8 May 2018 16:17:58 -0400 Subject: [PATCH] Fix RedisLockRegLeaderInitTests race condition https://build.spring.io/browse/INT-FATS5IC-501/ When we try to wait for the `Latch` in the interruptable code flow, it is a fact that we step away from the waiting and end up with the race condition downstream. * Wrap `Latch` in the interruptable `publishOnRevoked()` code to the `Executor.execute()` * Remove `deleteTimeoutMillis` option from the `RedisLockRegistry` since it doesn't make sense in the interruptable code. * Add `RedisLockRegistry.setExecutor()` to allow to inject an external `Executor` * Add more debug logging into the `LockRegistryLeaderInitiator` **Cherry-pick to 5.0.x** --- .../leader/LockRegistryLeaderInitiator.java | 22 ++++++--- .../redis/util/RedisLockRegistry.java | 35 ++++++------- ...RedisLockRegistryLeaderInitiatorTests.java | 49 +++++++++++++------ 3 files changed, 67 insertions(+), 39 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 0b068f0d44..fb2e503110 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 @@ -315,7 +315,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe this.future.cancel(true); } this.future = null; - logger.debug("Stopped LeaderInitiator"); + logger.debug("Stopped LeaderInitiator for " + getContext()); } } } @@ -347,6 +347,11 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe try { while (isRunning()) { try { + + if (logger.isDebugEnabled()) { + logger.debug("Acquiring the lock for " + this.context); + } + // We always try to acquire the lock, in case it expired boolean acquired = this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis, TimeUnit.MILLISECONDS); @@ -386,8 +391,8 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe this.lock.unlock(); } catch (Exception e1) { - logger.debug("Could not unlock - treat as broken: " + this.context + - ". Revoking " + (isRunning() ? " and retrying..." : "..."), e); + logger.debug("Could not unlock - treat as broken " + this.context + + ". Revoking " + (isRunning() ? " and retrying..." : "..."), e1); } // The lock was broken and we are no longer leader @@ -401,14 +406,15 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe if (e instanceof InterruptedException || Thread.currentThread().isInterrupted()) { Thread.currentThread().interrupt(); if (isRunning()) { - logger.warn("Restarting LeaderSelector because of error.", e); + logger.warn("Restarting LeaderSelector for " + this.context + " because of error.", e); LockRegistryLeaderInitiator.this.future = LockRegistryLeaderInitiator.this.executorService.submit(this); } return null; } else if (logger.isDebugEnabled()) { - logger.debug("Error acquiring the lock. " + (isRunning() ? "Retrying..." : ""), e); + logger.debug("Error acquiring the lock for " + this.context + + ". " + (isRunning() ? "Retrying..." : ""), e); } } } @@ -420,7 +426,8 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe this.lock.unlock(); } catch (Exception e) { - logger.debug("Could not unlock during stop - treat as broken. Revoking...", e); + logger.debug("Could not unlock during stop for " + this.context + + " - treat as broken. Revoking...", e); } // We are stopping, therefore not leading any more handleRevoked(); @@ -492,6 +499,9 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe @Override public void yield() { + if (logger.isDebugEnabled()) { + logger.debug("Yielding leadership from " + this); + } if (LockRegistryLeaderInitiator.this.future != null) { LockRegistryLeaderInitiator.this.future.cancel(true); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java index c192b090b0..c3be6e9c0d 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java @@ -23,6 +23,7 @@ import java.util.Iterator; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -79,8 +80,6 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl private static final long DEFAULT_EXPIRE_AFTER = 60000L; - public static final long DEFAULT_DELETE_TIMEOUT = 10000L; - private static final String OBTAIN_LOCK_SCRIPT = "local lockClientId = redis.call('GET', KEYS[1])\n" + "if lockClientId == ARGV[1] then\n" + @@ -97,21 +96,14 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl * An {@link ExecutorService} to call {@link StringRedisTemplate#delete(Object)} in * the separate thread when the current one is interrupted. */ - private ExecutorService executorService = + private Executor executor = Executors.newCachedThreadPool(new CustomizableThreadFactory("redis-lock-registry-")); /** * Flag to denote whether the {@link ExecutorService} was provided via the setter and * thus should not be shutdown when {@link #destroy()} is called */ - private boolean executorServiceExplicitlySet; - - /** - * Time in milliseconds to wait for the {@link StringRedisTemplate#delete(Object)} in - * the background thread. - */ - private long deleteTimeoutMillis = DEFAULT_DELETE_TIMEOUT; - + private boolean executorExplicitlySet; private final Map locks = new ConcurrentHashMap<>(); @@ -149,6 +141,17 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl this.expireAfter = expireAfter; } + /** + * Set the {@link Executor}, where is not provided then a default of + * cached thread pool Executor will be used. + * @param executor the executor service + * @since 5.0.5 + */ + public void setExecutor(Executor executor) { + this.executor = executor; + this.executorExplicitlySet = true; + } + @Override public Lock obtain(Object lockKey) { Assert.isInstanceOf(String.class, lockKey); @@ -171,8 +174,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl @Override public void destroy() { - if (!this.executorServiceExplicitlySet) { - this.executorService.shutdown(); + if (!this.executorExplicitlySet) { + ((ExecutorService) this.executor).shutdown(); } } @@ -298,11 +301,9 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl return; } try { - if (Thread.currentThread().isInterrupted()) { - RedisLockRegistry.this.executorService.submit(() -> - RedisLockRegistry.this.redisTemplate.delete(this.lockKey)) - .get(RedisLockRegistry.this.deleteTimeoutMillis, TimeUnit.MILLISECONDS); + RedisLockRegistry.this.executor.execute(() -> + RedisLockRegistry.this.redisTemplate.delete(this.lockKey)); } else { RedisLockRegistry.this.redisTemplate.delete(this.lockKey); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java index 65182de0ea..ba581a7ccd 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java @@ -23,8 +23,12 @@ import static org.junit.Assert.assertThat; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.junit.Rule; import org.junit.Test; @@ -36,6 +40,8 @@ import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.redis.util.RedisLockRegistry; import org.springframework.integration.support.leader.LockRegistryLeaderInitiator; import org.springframework.integration.test.rule.Log4j2LevelAdjuster; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; +import org.springframework.util.ReflectionUtils; /** * @author Artem Bilan @@ -46,10 +52,12 @@ import org.springframework.integration.test.rule.Log4j2LevelAdjuster; */ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests { + private static final Log logger = LogFactory.getLog(RedisLockRegistryLeaderInitiatorTests.class); + @Rule public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace() - .categories(true, "org.springframework.data.redis"); + .categories(true, "org.springframework.integration.redis.leader"); @Test @RedisAvailable @@ -62,6 +70,8 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests { for (int i = 0; i < 2; i++) { LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(registry, new DefaultCandidate("foo:" + i, "bar")); + initiator.setExecutorService( + Executors.newSingleThreadExecutor(new CustomizableThreadFactory("lock-leadership-" + i + "-"))); initiator.setLeaderEventPublisher(countingPublisher); initiators.add(initiator); } @@ -95,18 +105,23 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests { CountDownLatch acquireLockFailed1 = new CountDownLatch(1); CountDownLatch acquireLockFailed2 = new CountDownLatch(1); + Executor latchesExecutor = Executors.newCachedThreadPool(); + initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1, acquireLockFailed1) { @Override public void publishOnRevoked(Object source, Context context, String role) { - try { + latchesExecutor.execute(() -> { // It's difficult to see round-robin election, so block one initiator until the second is elected. - assertThat(granted2.await(10, TimeUnit.SECONDS), is(true)); - } - catch (InterruptedException e) { - // No op - } - super.publishOnRevoked(source, context, role); + try { + assertThat(granted2.await(10, TimeUnit.SECONDS), is(true)); + } + catch (InterruptedException e) { + ReflectionUtils.rethrowRuntimeException(e); + } + super.publishOnRevoked(source, context, role); + }); + } }); @@ -115,14 +130,16 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests { @Override public void publishOnRevoked(Object source, Context context, String role) { - try { - // It's difficult to see round-robin election, so block one initiator until the second is elected. - assertThat(granted1.await(10, TimeUnit.SECONDS), is(true)); - } - catch (InterruptedException e) { - // No op - } - super.publishOnRevoked(source, context, role); + latchesExecutor.execute(() -> { + try { + // It's difficult to see round-robin election, so block one initiator until the second is elected. + assertThat(granted1.await(10, TimeUnit.SECONDS), is(true)); + } + catch (InterruptedException e) { + ReflectionUtils.rethrowRuntimeException(e); + } + super.publishOnRevoked(source, context, role); + }); } });