Merge branch 'stable-3.0' into stable-3.1
* stable-3.0:
Remove index.waitTimeout configuration property
Fix indexing locks on the sender side
Align Project.NameKey serialization to gerrit core
Lock on index element id rather than on index task id
Do not change Gerrit files ownership
Change-Id: I5df0768de62fe3bbbfae18be18442cbf7c09d1ec
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java
index 3bd58a3..4743bee 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java
@@ -455,7 +455,6 @@
private final int threadPoolSize;
private final int retryInterval;
private final int maxTries;
- private final int waitTimeout;
private final int numStripedLocks;
private final boolean synchronizeForced;
@@ -465,7 +464,6 @@
numStripedLocks = getInt(cfg, INDEX_SECTION, NUM_STRIPED_LOCKS, DEFAULT_NUM_STRIPED_LOCKS);
retryInterval = getInt(cfg, INDEX_SECTION, RETRY_INTERVAL_KEY, DEFAULT_INDEX_RETRY_INTERVAL);
maxTries = getInt(cfg, INDEX_SECTION, MAX_TRIES_KEY, DEFAULT_INDEX_MAX_TRIES);
- waitTimeout = getInt(cfg, INDEX_SECTION, WAIT_TIMEOUT_KEY, DEFAULT_TIMEOUT_MS);
synchronizeForced =
cfg.getBoolean(INDEX_SECTION, SYNCHRONIZE_FORCED_KEY, DEFAULT_SYNCHRONIZE_FORCED);
}
@@ -486,10 +484,6 @@
return maxTries;
}
- public int waitTimeout() {
- return waitTimeout;
- }
-
public boolean synchronizeForced() {
return synchronizeForced;
}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java
index b311db9..368581c 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java
@@ -171,6 +171,8 @@
}
abstract CompletableFuture<Boolean> execute();
+
+ abstract String indexId();
}
class IndexChangeTask extends IndexTask {
@@ -206,6 +208,11 @@
public String toString() {
return String.format("[%s] Index change %s in target instance", pluginName, changeId);
}
+
+ @Override
+ String indexId() {
+ return "change/" + changeId;
+ }
}
class DeleteChangeTask extends IndexTask {
@@ -239,6 +246,11 @@
public String toString() {
return String.format("[%s] Delete change %s in target instance", pluginName, changeId);
}
+
+ @Override
+ String indexId() {
+ return "change/" + changeId;
+ }
}
class IndexAccountTask extends IndexTask {
@@ -271,6 +283,11 @@
public String toString() {
return String.format("[%s] Index account %s in target instance", pluginName, accountId);
}
+
+ @Override
+ String indexId() {
+ return "account/" + accountId;
+ }
}
class IndexGroupTask extends IndexTask {
@@ -303,6 +320,11 @@
public String toString() {
return String.format("[%s] Index group %s in target instance", pluginName, groupUUID);
}
+
+ @Override
+ String indexId() {
+ return "group/" + groupUUID;
+ }
}
class IndexProjectTask extends IndexTask {
@@ -335,5 +357,10 @@
public String toString() {
return String.format("[%s] Index project %s in target instance", pluginName, projectName);
}
+
+ @Override
+ String indexId() {
+ return "project/" + projectName;
+ }
}
}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventLocks.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventLocks.java
index 4434c34..3fd26db 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventLocks.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventLocks.java
@@ -21,45 +21,66 @@
import com.google.common.util.concurrent.Striped;
import com.google.inject.Inject;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;
public class IndexEventLocks {
private static final FluentLogger log = FluentLogger.forEnclosingClass();
private static final int NUMBER_OF_INDEX_TASK_TYPES = 4;
+ private static final int WAIT_TIMEOUT_MS = 5;
- private final Striped<Lock> locks;
- private final long waitTimeout;
+ private final Striped<Semaphore> semaphores;
@Inject
public IndexEventLocks(Configuration cfg) {
- this.locks = Striped.lock(NUMBER_OF_INDEX_TASK_TYPES * cfg.index().numStripedLocks());
- this.waitTimeout = cfg.index().waitTimeout();
+ this.semaphores =
+ Striped.semaphore(NUMBER_OF_INDEX_TASK_TYPES * cfg.index().numStripedLocks(), 1);
}
- public void withLock(
+ public CompletableFuture<?> withLock(
IndexTask id, IndexCallFunction function, VoidFunction lockAcquireTimeoutCallback) {
- Lock idLock = getLock(id);
+ String indexId = id.indexId();
+ Semaphore idSemaphore = getSemaphore(indexId);
try {
- if (idLock.tryLock(waitTimeout, TimeUnit.MILLISECONDS)) {
- function
+ log.atFine().log("Trying to acquire %s", id);
+ if (idSemaphore.tryAcquire(WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
+ log.atFine().log("Acquired %s", id);
+ return function
.invoke()
.whenComplete(
(result, error) -> {
- idLock.unlock();
+ try {
+ log.atFine().log("Trying to release %s", id);
+ idSemaphore.release();
+ log.atFine().log("Released %s", id);
+ } catch (Throwable t) {
+ log.atSevere().withCause(t).log("Unable to release %s", id);
+ throw t;
+ }
});
- } else {
- lockAcquireTimeoutCallback.invoke();
}
+
+ String timeoutMessage =
+ String.format(
+ "Acquisition of the locking of %s timed out after %d msec: consider increasing the number of shards",
+ indexId, WAIT_TIMEOUT_MS);
+ log.atWarning().log(timeoutMessage);
+ lockAcquireTimeoutCallback.invoke();
+ CompletableFuture<?> failureFuture = new CompletableFuture<>();
+ failureFuture.completeExceptionally(new InterruptedException(timeoutMessage));
+ return failureFuture;
} catch (InterruptedException e) {
- log.atSevere().withCause(e).log("%s was interrupted; giving up", id);
+ CompletableFuture<?> failureFuture = new CompletableFuture<>();
+ failureFuture.completeExceptionally(e);
+ log.atSevere().withCause(e).log("Locking of %s was interrupted; giving up", indexId);
+ return failureFuture;
}
}
@VisibleForTesting
- protected Lock getLock(IndexTask id) {
- return locks.get(id);
+ protected Semaphore getSemaphore(String indexId) {
+ return semaphores.get(indexId);
}
@FunctionalInterface
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index e4055cb..68fa8eb 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -207,13 +207,8 @@
: The interval of time in milliseconds between the subsequent auto-retries.
Defaults to 30000 (30 seconds).
-```index.waitTimeout```
-: Maximum interval of time in milliseconds the plugin waits to acquire
- the lock for an indexing call. When not specified, the default value
- is set to 5000ms.
-
-NOTE: the default settings for `http.socketTimeout`, `http.maxTries` and `index.waitTimeout`
-ensure that the plugin will keep retrying to forward a message for one hour.
+NOTE: the default settings for `http.socketTimeout` and `http.maxTries` ensure
+that the plugin will keep retrying to forward a message for one hour.
```websession.synchronize```
: Whether to synchronize web sessions.
diff --git a/src/test/docker/gerrit/start.sh b/src/test/docker/gerrit/start.sh
index 65804b3..0dae4c3 100755
--- a/src/test/docker/gerrit/start.sh
+++ b/src/test/docker/gerrit/start.sh
@@ -5,8 +5,6 @@
wait-for-it.sh $WAIT_FOR -t 600 -- echo "$WAIT_FOR is up"
fi
-chown -R gerrit: /var/gerrit
-
sudo -u gerrit cp /var/gerrit/etc/gerrit.config.orig /var/gerrit/etc/gerrit.config
sudo -u gerrit cp /var/gerrit/etc/high-availability.config.orig /var/gerrit/etc/high-availability.config
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java
index db4c613..875f203 100644
--- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java
+++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java
@@ -16,6 +16,7 @@
import static com.google.common.truth.Truth.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -32,7 +33,8 @@
import com.ericsson.gerrit.plugins.highavailability.index.IndexEventHandler.IndexAccountTask;
import com.ericsson.gerrit.plugins.highavailability.index.IndexEventHandler.IndexChangeTask;
import com.ericsson.gerrit.plugins.highavailability.index.IndexEventHandler.IndexGroupTask;
-import com.ericsson.gerrit.plugins.highavailability.index.IndexEventHandler.IndexProjectTask;
+import com.ericsson.gerrit.plugins.highavailability.index.IndexEventHandler.IndexTask;
+import com.ericsson.gerrit.plugins.highavailability.index.IndexEventLocks.VoidFunction;
import com.google.gerrit.entities.Account;
import com.google.gerrit.entities.AccountGroup;
import com.google.gerrit.entities.Change;
@@ -44,15 +46,19 @@
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.locks.Lock;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
+import java.util.function.Supplier;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -68,6 +74,8 @@
private static final int ACCOUNT_ID = 2;
private static final String UUID = "3";
private static final String OTHER_UUID = "4";
+ private static final Integer INDEX_WAIT_TIMEOUT_MS = 5;
+ private static final int MAX_TEST_PARALLELISM = 4;
private IndexEventHandler indexEventHandler;
@Mock private Forwarder forwarder;
@@ -77,6 +85,8 @@
private Account.Id accountId;
private AccountGroup.UUID accountGroupUUID;
private ScheduledExecutorService executor = new CurrentThreadScheduledExecutorService();
+ private ScheduledExecutorService testExecutor =
+ Executors.newScheduledThreadPool(MAX_TEST_PARALLELISM);
@Mock private RequestContext mockCtx;
@Mock private Configuration configuration;
private IndexEventLocks idLocks;
@@ -100,7 +110,6 @@
Configuration.Index cfgIndex = mock(Configuration.Index.class);
when(configuration.index()).thenReturn(cfgIndex);
when(cfgIndex.numStripedLocks()).thenReturn(Configuration.DEFAULT_NUM_STRIPED_LOCKS);
- when(cfgIndex.waitTimeout()).thenReturn(Configuration.DEFAULT_TIMEOUT_MS);
Configuration.Http http = mock(Configuration.Http.class);
when(configuration.http()).thenReturn(http);
@@ -163,10 +172,11 @@
@Test
public void shouldNotIndexChangeWhenCannotAcquireLock() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock lock = mock(Lock.class);
- when(locks.getLock(any(IndexChangeTask.class))).thenReturn(lock);
+ Semaphore semaphore = mock(Semaphore.class);
+ when(locks.getSemaphore(anyString())).thenReturn(semaphore);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
- when(lock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS))).thenReturn(false);
+ when(semaphore.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(false);
setUpIndexEventHandler(currCtx, locks);
indexEventHandler.onChangeIndexed(PROJECT_NAME, changeId.get());
@@ -177,9 +187,10 @@
@Test
public void shouldNotIndexAccountWhenCannotAcquireLock() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock lock = mock(Lock.class);
- when(locks.getLock(any(IndexAccountTask.class))).thenReturn(lock);
- when(lock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS))).thenReturn(false);
+ Semaphore semaphore = mock(Semaphore.class);
+ when(locks.getSemaphore(anyString())).thenReturn(semaphore);
+ when(semaphore.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(false);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
setUpIndexEventHandler(currCtx, locks);
@@ -191,9 +202,10 @@
@Test
public void shouldNotDeleteChangeWhenCannotAcquireLock() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock lock = mock(Lock.class);
- when(locks.getLock(any(DeleteChangeTask.class))).thenReturn(lock);
- when(lock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS))).thenReturn(false);
+ Semaphore semaphore = mock(Semaphore.class);
+ when(locks.getSemaphore(anyString())).thenReturn(semaphore);
+ when(semaphore.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(false);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
setUpIndexEventHandler(currCtx, locks);
@@ -205,9 +217,10 @@
@Test
public void shouldNotIndexGroupWhenCannotAcquireLock() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock lock = mock(Lock.class);
- when(locks.getLock(any(IndexGroupTask.class))).thenReturn(lock);
- when(lock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS))).thenReturn(false);
+ Semaphore semaphore = mock(Semaphore.class);
+ when(locks.getSemaphore(anyString())).thenReturn(semaphore);
+ when(semaphore.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(false);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
setUpIndexEventHandler(currCtx, locks);
@@ -219,9 +232,10 @@
@Test
public void shouldNotIndexProjectWhenCannotAcquireLock() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock lock = mock(Lock.class);
- when(locks.getLock(any(IndexProjectTask.class))).thenReturn(lock);
- when(lock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS))).thenReturn(false);
+ Semaphore semaphore = mock(Semaphore.class);
+ when(locks.getSemaphore(anyString())).thenReturn(semaphore);
+ when(semaphore.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(false);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
setUpIndexEventHandler(currCtx, locks);
@@ -233,10 +247,10 @@
@Test
public void shouldRetryIndexChangeWhenCannotAcquireLock() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock lock = mock(Lock.class);
- when(locks.getLock(any(IndexChangeTask.class))).thenReturn(lock);
+ Semaphore semaphore = mock(Semaphore.class);
+ when(locks.getSemaphore(anyString())).thenReturn(semaphore);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
- when(lock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ when(semaphore.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
.thenReturn(false, true);
setUpIndexEventHandler(currCtx, locks);
@@ -249,10 +263,11 @@
@Test
public void shouldRetryUpToMaxTriesWhenCannotAcquireLock() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock lock = mock(Lock.class);
- when(locks.getLock(any(IndexChangeTask.class))).thenReturn(lock);
+ Semaphore semaphore = mock(Semaphore.class);
+ when(locks.getSemaphore(anyString())).thenReturn(semaphore);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
- when(lock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS))).thenReturn(false);
+ when(semaphore.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(false);
Configuration cfg = mock(Configuration.class);
Configuration.Http httpCfg = mock(Configuration.Http.class);
@@ -269,10 +284,11 @@
@Test
public void shouldNotRetryWhenMaxTriesLowerThanOne() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock lock = mock(Lock.class);
- when(locks.getLock(any(IndexChangeTask.class))).thenReturn(lock);
+ Semaphore semaphore = mock(Semaphore.class);
+ when(locks.getSemaphore(anyString())).thenReturn(semaphore);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
- when(lock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS))).thenReturn(false);
+ when(semaphore.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(false);
Configuration cfg = mock(Configuration.class);
Configuration.Http httpCfg = mock(Configuration.Http.class);
@@ -289,14 +305,14 @@
@Test
public void shouldLockPerIndexEventType() throws Exception {
IndexEventLocks locks = mock(IndexEventLocks.class);
- Lock indexChangeLock = mock(Lock.class);
- when(indexChangeLock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ Semaphore indexChangeLock = mock(Semaphore.class);
+ when(indexChangeLock.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
.thenReturn(false);
- Lock accountChangeLock = mock(Lock.class);
- when(accountChangeLock.tryLock(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
+ Semaphore accountChangeLock = mock(Semaphore.class);
+ when(accountChangeLock.tryAcquire(Mockito.anyLong(), Mockito.eq(TimeUnit.MILLISECONDS)))
.thenReturn(true);
- when(locks.getLock(any(IndexChangeTask.class))).thenReturn(indexChangeLock);
- when(locks.getLock(any(IndexAccountTask.class))).thenReturn(accountChangeLock);
+ when(locks.getSemaphore(eq("change/" + CHANGE_ID))).thenReturn(indexChangeLock);
+ when(locks.getSemaphore(eq("account/" + ACCOUNT_ID))).thenReturn(accountChangeLock);
Mockito.doCallRealMethod().when(locks).withLock(any(), any(), any());
setUpIndexEventHandler(currCtx, locks);
@@ -531,6 +547,134 @@
assertThat(task.hashCode()).isNotEqualTo(differentGroupIdTask.hashCode());
}
+ class TestTask<T> implements Runnable {
+ private IndexTask task;
+ private CyclicBarrier testBarrier;
+ private Supplier<T> successFunc;
+ private VoidFunction failureFunc;
+ private CompletableFuture<T> future;
+
+ public TestTask(
+ IndexTask task,
+ CyclicBarrier testBarrier,
+ Supplier<T> successFunc,
+ VoidFunction failureFunc) {
+ this.task = task;
+ this.testBarrier = testBarrier;
+ this.successFunc = successFunc;
+ this.failureFunc = failureFunc;
+ this.future = new CompletableFuture<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void run() {
+ try {
+ testBarrier.await();
+ idLocks
+ .withLock(
+ task,
+ () ->
+ runLater(
+ INDEX_WAIT_TIMEOUT_MS * 2,
+ () -> CompletableFuture.completedFuture(successFunc.get())),
+ failureFunc)
+ .whenComplete(
+ (v, t) -> {
+ if (t == null) {
+ future.complete((T) v);
+ } else {
+ future.completeExceptionally(t);
+ }
+ });
+ } catch (Throwable t) {
+ future = new CompletableFuture<>();
+ future.completeExceptionally(t);
+ }
+ }
+
+ public void join() {
+ try {
+ future.join();
+ } catch (Exception e) {
+ }
+ }
+
+ private CompletableFuture<T> runLater(
+ long scheduledTimeMsec, Supplier<CompletableFuture<T>> supplier) {
+ CompletableFuture<T> resFuture = new CompletableFuture<>();
+ testExecutor.schedule(
+ () -> {
+ try {
+ return supplier
+ .get()
+ .whenComplete(
+ (v, t) -> {
+ if (t == null) {
+ resFuture.complete(v);
+ }
+ resFuture.completeExceptionally(t);
+ });
+ } catch (Throwable t) {
+ return resFuture.completeExceptionally(t);
+ }
+ },
+ scheduledTimeMsec,
+ TimeUnit.MILLISECONDS);
+ return resFuture;
+ }
+ }
+
+ @Test
+ public void indexLocksShouldBlockConcurrentIndexChange() throws Exception {
+ IndexChangeTask indexTask1 =
+ indexEventHandler.new IndexChangeTask(PROJECT_NAME, CHANGE_ID, new IndexEvent());
+ IndexChangeTask indexTask2 =
+ indexEventHandler.new IndexChangeTask(PROJECT_NAME, CHANGE_ID, new IndexEvent());
+ testIsolationOfCuncurrentIndexTasks(indexTask1, indexTask2);
+ }
+
+ @Test
+ public void indexLocksShouldBlockConcurrentIndexAndDeleteChange() throws Exception {
+ IndexChangeTask indexTask =
+ indexEventHandler.new IndexChangeTask(PROJECT_NAME, CHANGE_ID, new IndexEvent());
+ DeleteChangeTask deleteTask =
+ indexEventHandler.new DeleteChangeTask(CHANGE_ID, new IndexEvent());
+ testIsolationOfCuncurrentIndexTasks(indexTask, deleteTask);
+ }
+
+ private void testIsolationOfCuncurrentIndexTasks(IndexTask indexTask1, IndexTask indexTask2)
+ throws Exception {
+ AtomicInteger changeIndexedCount = new AtomicInteger();
+ AtomicInteger lockFailedCounts = new AtomicInteger();
+ CyclicBarrier changeThreadsSync = new CyclicBarrier(2);
+
+ TestTask<Integer> task1 =
+ new TestTask<>(
+ indexTask1,
+ changeThreadsSync,
+ () -> changeIndexedCount.incrementAndGet(),
+ () -> lockFailedCounts.incrementAndGet());
+ TestTask<Integer> task2 =
+ new TestTask<>(
+ indexTask2,
+ changeThreadsSync,
+ () -> changeIndexedCount.incrementAndGet(),
+ () -> lockFailedCounts.incrementAndGet());
+
+ new Thread(task1).start();
+ new Thread(task2).start();
+ task1.join();
+ task2.join();
+
+ /* Both assertions needs to be true, the order doesn't really matter:
+ * - Only one of the two tasks should succeed
+ * - Only one of the two tasks should fail to acquire the lock
+ */
+ assertThat(changeIndexedCount.get()).isEqualTo(1);
+ assertThat(lockFailedCounts.get()).isEqualTo(1);
+ }
+
private class CurrentThreadScheduledExecutorService implements ScheduledExecutorService {
@Override
