Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix RedisLockRegLeaderInitTests race condition

Browse Source 
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**
This commit is contained in:
Artem Bilan
2018-05-08 16:17:58 -04:00
parent 3c1b547b26
commit 51c49519dc
3 changed files with 67 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java
View File

@@ -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);
}

35
spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
View File

@@ -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<String, RedisLock> 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);

49
spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java
View File

@@ -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);
});
}
});