libre.sh/keycloak-scim
3
0
Fork 0
Code Issues 15 Pull requests Releases Packages Activity Actions

Fix testReleaseAllLocksMethod timing out intermittently

Browse source 
Closes #17337
This commit is contained in:
Michal Hajas 2023-03-01 12:22:49 +01:00 committed by Hynek Mlnařík
parent 4c4015cf0b
commit e02c95f9d3
1 changed files with 64 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

112
testsuite/model/src/test/java/org/keycloak/testsuite/model/globalLock/GlobalLocksTest.java
View file

@ -34,6 +34,7 @@ import java.util.Random;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import java.util.stream.IntStream; import java.util.stream.IntStream;
@ -128,73 +129,88 @@ public class GlobalLocksTest extends KeycloakModelTest {
@Test @Test
public void testReleaseAllLocksMethod() throws InterruptedException { public void testReleaseAllLocksMethod() throws InterruptedException {
final int NUMBER_OF_THREADS = 4; final int numberOfThreads = 4;
ExecutorService executor = Executors.newFixedThreadPool(NUMBER_OF_THREADS); ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
CountDownLatch locksAcquired = new CountDownLatch(NUMBER_OF_THREADS); CountDownLatch locksAcquired = new CountDownLatch(numberOfThreads);
CountDownLatch testFinished = new CountDownLatch(1); CountDownLatch testFinished = new CountDownLatch(1);
LOG.info("Initial locks acquiring phase.");
try { try {
// Acquire locks and let the threads wait until the end of this test method // Acquire locks and let the threads wait until the end of this test method
executor.submit(() -> { for (int index = 0; index < numberOfThreads; index++) {
IntStream.range(0, NUMBER_OF_THREADS).parallel() final int i = index;
.forEach(i -> executor.submit(() ->
inComittedTransaction(s -> { inComittedTransaction(s -> {
GlobalLockProvider lockProvider = s.getProvider(GlobalLockProvider.class); GlobalLockProvider lockProvider = s.getProvider(GlobalLockProvider.class);
lockProvider.withLock("LOCK_" + i, session -> { LOG.infof("Acquiring LOCK_%d", i);
locksAcquired.countDown(); lockProvider.withLock("LOCK_" + i, session -> {
try { LOG.infof("Lock LOCK_%d acquired.", i);
testFinished.await(); locksAcquired.countDown();
} catch (InterruptedException e) { try {
Thread.currentThread().interrupt(); testFinished.await();
throw new RuntimeException(e); } catch (InterruptedException e) {
} Thread.currentThread().interrupt();
return null; throw new RuntimeException(e);
}); }
}) return null;
); });
}); LOG.infof("Initial acquiring tx finished for lock LOCK_%d", i);
})
);
}
locksAcquired.await(); if (!locksAcquired.await(5, TimeUnit.MINUTES)) {
throw new RuntimeException("Acquiring locks phase took too long.");
}
LOG.info("Expecting timeouts for each lock.");
// Test no lock can be acquired because all are still hold by the executor above // Test no lock can be acquired because all are still hold by the executor above
AtomicInteger counter = new AtomicInteger(); AtomicInteger counter = new AtomicInteger();
IntStream.range(0, NUMBER_OF_THREADS).parallel() for (int index = 0; index < numberOfThreads; index++) {
.forEach(i -> final int i = index;
inComittedTransaction(s -> { inComittedTransaction(s -> {
GlobalLockProvider lockProvider = s.getProvider(GlobalLockProvider.class); GlobalLockProvider lockProvider = s.getProvider(GlobalLockProvider.class);
try { try {
lockProvider.withLock("LOCK_" + i, Duration.ofSeconds(1), is -> { LOG.infof("Attempt to acquire LOCK_%d.", i);
throw new RuntimeException("Acquiring lock should not succeed as it was acquired in the first transaction"); lockProvider.withLock("LOCK_" + i, Duration.ofSeconds(1), is -> {
}); throw new RuntimeException("Acquiring lock should not succeed as it was acquired in the first transaction");
} catch (LockAcquiringTimeoutException e) { });
counter.incrementAndGet(); } catch (LockAcquiringTimeoutException e) {
} LOG.infof("Timeout was successfully received for LOCK_%d", i);
}) counter.incrementAndGet();
); }
assertThat(counter.get(), Matchers.equalTo(NUMBER_OF_THREADS)); });
}
assertThat(counter.get(), Matchers.equalTo(numberOfThreads));
// Unlock all locks forcefully // Unlock all locks forcefully
inComittedTransaction(s -> { inComittedTransaction(s -> {
GlobalLockProvider lockProvider = s.getProvider(GlobalLockProvider.class); GlobalLockProvider lockProvider = s.getProvider(GlobalLockProvider.class);
LOG.infof("Releasing all locks", Thread.currentThread().getName());
lockProvider.forceReleaseAllLocks(); lockProvider.forceReleaseAllLocks();
}); });
// Test all locks can be acquired again // Test all locks can be acquired again
counter.set(0); counter.set(0);
IntStream.range(0, NUMBER_OF_THREADS).parallel() for (int index = 0; index < numberOfThreads; index++) {
.forEach(i -> final int i = index;
inComittedTransaction(s -> { inComittedTransaction(s -> {
GlobalLockProvider lockProvider = s.getProvider(GlobalLockProvider.class); GlobalLockProvider lockProvider = s.getProvider(GlobalLockProvider.class);
try { try {
lockProvider.withLock("LOCK_" + i, Duration.ofSeconds(1), is -> counter.incrementAndGet()); lockProvider.withLock("LOCK_" + i, Duration.ofSeconds(1), is -> {
} catch (LockAcquiringTimeoutException e) { LOG.infof("Lock LOCK_%d acquired again.", i);
throw new RuntimeException(e); counter.incrementAndGet();
} return null;
}) });
); } catch (LockAcquiringTimeoutException e) {
throw new RuntimeException(e);
}
});
}
assertThat(counter.get(), Matchers.equalTo(NUMBER_OF_THREADS)); assertThat(counter.get(), Matchers.equalTo(numberOfThreads));
} finally { } finally {
testFinished.countDown(); testFinished.countDown();
executor.shutdown(); executor.shutdown();