keith-turner commented on code in PR #4524:
URL: https://github.com/apache/accumulo/pull/4524#discussion_r1631616426
##########
core/src/main/java/org/apache/accumulo/core/fate/MetaFateStore.java:
##########
@@ -315,6 +420,30 @@ public List<ReadOnlyRepo<T>> getStack() {
return dops;
}
}
+
+ @Override
+ protected void unreserve() {
+ try {
+ if (!this.deleted) {
+ zk.mutateExisting(getTXPath(fateId), currSerNodeVal -> {
+ NodeValue currNodeVal = new NodeValue(currSerNodeVal);
+ FateKey currFateKey = currNodeVal.fateKey.orElse(null);
+ if ((currNodeVal.isReserved()
+ &&
currNodeVal.reservation.orElseThrow().equals(this.reservation))) {
+ // Remove the FateReservation from the NodeValue to unreserve
+ return new NodeValue(currNodeVal.status, null,
currFateKey).serialize();
+ } else {
+ // possible this is running a 2nd time in zk server fault
conditions and its first
Review Comment:
Could add some logging here. If the reservation in ZK is not set, then
could log something at trace. If there is a reservation in ZK that differs, it
could be log something at debug. These may be useful later for debugging
problems when initially using the elasticity changes.
##########
core/src/main/java/org/apache/accumulo/core/fate/MetaFateStore.java:
##########
@@ -421,14 +548,30 @@ private Optional<FateKey>
deserializeFateKey(DataInputBuffer buffer) throws IOEx
return Optional.empty();
}
+ private Optional<FateReservation>
deserializeFateReservation(DataInputBuffer buffer)
+ throws IOException {
+ int length = buffer.readInt();
+ if (length > 0) {
+ return
Optional.of(FateReservation.deserialize(buffer.readNBytes(length)));
+ }
+ return Optional.empty();
+ }
+
byte[] serialize() {
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos)) {
dos.writeUTF(status.name());
+ if (isReserved()) {
Review Comment:
Not a change to make in this PR. Thinking if we push all of the Fate data
into a single node it would be nice to use JSON. That would make the data
stored in ZK more human readable and it would make it easier to serialize and
de-serialize.
##########
core/src/main/java/org/apache/accumulo/core/fate/MetaFateStore.java:
##########
@@ -224,9 +323,14 @@ public void setStatus(TStatus status) {
verifyReserved(true);
try {
- zk.putPersistentData(getTXPath(fateId), new
NodeValue(status).serialize(),
- NodeExistsPolicy.OVERWRITE);
- } catch (KeeperException | InterruptedException e) {
+ zk.mutateExisting(getTXPath(fateId), currSerializedData -> {
+ NodeValue currNodeVal = new NodeValue(currSerializedData);
Review Comment:
Somewhere in this lamba it should check a reservation exist for the ZK node
and that the reservation on the ZK node is equal to
AbstractFateTxStoreImpl.reserveration, if those conditions are not met then
throw an exception. This check ensures that the expected state we have in
memory aligns with what is in the persistent store.
Ideally these kinds of checks would be done for all updates, however it is
not possible to atomically do these checks for data stored in other ZK nodes.
The way FATE stores data in ZK it uses multiple nodes. The schema could be
changed to store all data in a single node which would allow all updates to
verify the reservation atomically, that is not a change for this PR. Would be
a follow on issue.
##########
core/src/main/java/org/apache/accumulo/core/fate/MetaFateStore.java:
##########
@@ -106,13 +112,102 @@ public FateId create() {
@Override
protected void create(FateId fateId, FateKey key) {
try {
- zk.putPersistentData(getTXPath(fateId), new NodeValue(TStatus.NEW,
key).serialize(),
+ zk.putPersistentData(getTXPath(fateId), new NodeValue(TStatus.NEW, null,
key).serialize(),
NodeExistsPolicy.FAIL);
} catch (KeeperException | InterruptedException e) {
throw new IllegalStateException(e);
}
}
+ @Override
+ public Optional<FateTxStore<T>> tryReserve(FateId fateId) {
+ // uniquely identify this attempt to reserve the fate operation data
+ FateReservation reservation = FateReservation.from(lockID,
UUID.randomUUID());
+
+ try {
+ byte[] newSerNodeVal = zk.mutateExisting(getTXPath(fateId),
currSerNodeVal -> {
+ NodeValue currNodeVal = new NodeValue(currSerNodeVal);
+ // The uuid handles the case where there was a ZK server fault and the
write for this thread
+ // went through but that was not acknowledged, and we are reading our
own write for 2nd
+ // time.
+ if (!currNodeVal.isReserved() || (currNodeVal.isReserved()
+ && currNodeVal.reservation.orElseThrow().equals(reservation))) {
+ FateKey currFateKey = currNodeVal.fateKey.orElse(null);
+ // Add the FateReservation to the node to reserve
+ return new NodeValue(currNodeVal.status, reservation,
currFateKey).serialize();
+ } else {
+ // This will not change the value to null but will return null
+ return null;
+ }
+ });
+ if (newSerNodeVal != null) {
+ return Optional.of(new FateTxStoreImpl(fateId, reservation));
+ } else {
+ return Optional.empty();
+ }
+ } catch (InterruptedException | KeeperException |
AcceptableThriftTableOperationException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public boolean isReserved(FateId fateId) {
+ boolean isReserved;
+ try {
+ isReserved = getNode(fateId).isReserved();
+ } catch (Exception e) {
+ // Exception thrown, so node doesn't exist, so it is not reserved
+ isReserved = false;
+ }
Review Comment:
If possible this catch should be more narrow. Only want to ignore
exceptions that would be expected for the case of it not existing. Other
exceptions should bubble up so bugs can be detected. Also nice to log
something when ignoring an exception.
```suggestion
} catch (NoNodeException e) {
log.trace("Saw {} when checking reservation of {}", e.getMessage(),
fateId);
isReserved = false;
} catch (Exception e) {
throw new RuntimeException(e);
}
```
##########
test/src/main/java/org/apache/accumulo/test/fate/MultipleStoresIT.java:
##########
@@ -0,0 +1,571 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.fate;
+
+import static
org.apache.accumulo.test.fate.user.UserFateStoreIT.createFateTable;
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.conf.DefaultConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.FateId;
+import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.fate.FateStore;
+import org.apache.accumulo.core.fate.MetaFateStore;
+import org.apache.accumulo.core.fate.ReadOnlyFateStore;
+import org.apache.accumulo.core.fate.Repo;
+import org.apache.accumulo.core.fate.user.UserFateStore;
+import org.apache.accumulo.core.fate.zookeeper.ZooCache;
+import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.util.Wait;
+import org.apache.accumulo.test.zookeeper.ZooKeeperTestingServer;
+import org.easymock.EasyMock;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Sets;
+
+// TODO 4131 could potentially have separate classes for testing MetaFateStore
and UserFateStore
+// similar to how FateTestRunner is used, however that interface doesn't work
as nicely here
+// since we are using multiple stores instead of just one. Can do something
similar to
+// FateTestRunner here if desired
+public class MultipleStoresIT extends SharedMiniClusterBase {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MultipleStoresIT.class);
+ @TempDir
+ private static File tempDir;
+ private static ZooKeeperTestingServer szk = null;
+ private static ZooReaderWriter zk;
+ private static final String FATE_DIR = "/fate";
+ private ClientContext client;
+
+ @BeforeEach
+ public void beforeEachSetup() {
+ client = (ClientContext)
Accumulo.newClient().from(getClientProps()).build();
+ }
+
+ @AfterEach
+ public void afterEachTeardown() {
+ client.close();
+ }
+
+ @BeforeAll
+ public static void beforeAllSetup() throws Exception {
+ SharedMiniClusterBase.startMiniCluster();
+ szk = new ZooKeeperTestingServer(tempDir);
+ zk = szk.getZooReaderWriter();
+ }
+
+ @AfterAll
+ public static void afterAllTeardown() throws Exception {
+ SharedMiniClusterBase.stopMiniCluster();
+ szk.close();
+ }
+
+ @Test
+ public void testReserveUnreserve() throws Exception {
+ testReserveUnreserve(FateInstanceType.META);
+ testReserveUnreserve(FateInstanceType.USER);
+ }
+
+ private void testReserveUnreserve(FateInstanceType storeType) throws
Exception {
+ // reserving/unreserving a FateId should be reflected across instances of
the stores
+ final String tableName = getUniqueNames(1)[0];
+ final int numFateIds = 500;
+ final List<FateStore.FateTxStore<SleepingTestEnv>> reservations = new
ArrayList<>();
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final FateStore<SleepingTestEnv> store1, store2;
+ final ZooUtil.LockID lock1 = new ZooUtil.LockID("/locks", "L1", 50);
+ final ZooUtil.LockID lock2 = new ZooUtil.LockID("/locks", "L2", 52);
+ Map<FateId,FateStore.FateReservation> activeReservations;
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store1 = new UserFateStore<>(client, tableName, lock1);
+ store2 = new UserFateStore<>(client, tableName, lock2);
+ } else {
+ store1 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock1);
+ store2 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock2);
+ }
+
+ // Create the fate ids using store1
+ for (int i = 0; i < numFateIds; i++) {
+ assertTrue(allIds.add(store1.create()));
+ }
+ assertEquals(numFateIds, allIds.size());
+
+ // Reserve half the fate ids using store1 and rest using store2, after
reserving a fate id in
+ // one, should not be able to reserve the same in the other. Should also
not matter that all the
+ // ids were created using store1
+ int count = 0;
+ for (FateId fateId : allIds) {
+ if (count % 2 == 0) {
+ reservations.add(store1.reserve(fateId));
+ assertTrue(store2.tryReserve(fateId).isEmpty());
+ } else {
+ reservations.add(store2.reserve(fateId));
+ assertTrue(store1.tryReserve(fateId).isEmpty());
+ }
+ count++;
+ }
+ // Both stores should return the same reserved transactions
+ activeReservations = store1.getActiveReservations();
+ assertEquals(allIds, activeReservations.keySet());
+ activeReservations = store2.getActiveReservations();
+ assertEquals(allIds, activeReservations.keySet());
+
+ // Test setting/getting the TStatus and unreserving the transactions
+ for (int i = 0; i < allIds.size(); i++) {
+ var reservation = reservations.get(i);
+ assertEquals(ReadOnlyFateStore.TStatus.NEW, reservation.getStatus());
+ reservation.setStatus(ReadOnlyFateStore.TStatus.SUBMITTED);
+ assertEquals(ReadOnlyFateStore.TStatus.SUBMITTED,
reservation.getStatus());
+ reservation.delete();
+ reservation.unreserve(Duration.ofMillis(0));
+ // Attempt to set a status on a tx that has been unreserved (should
throw exception)
+ assertThrows(IllegalStateException.class,
+ () -> reservation.setStatus(ReadOnlyFateStore.TStatus.NEW));
+ }
+ assertTrue(store1.getActiveReservations().isEmpty());
+ assertTrue(store2.getActiveReservations().isEmpty());
+ }
+
+ @Test
+ public void testReserveNonExistentTxn() throws Exception {
+ testReserveNonExistentTxn(FateInstanceType.META);
+ testReserveNonExistentTxn(FateInstanceType.USER);
+ }
+
+ private void testReserveNonExistentTxn(FateInstanceType storeType) throws
Exception {
+ // Tests that reserve() doesn't hang indefinitely and instead throws an
error
+ // on reserve() a non-existent transaction.
+ final FateStore<SleepingTestEnv> store;
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final String tableName = getUniqueNames(1)[0];
+ final FateId fakeFateId = FateId.from(storeType, UUID.randomUUID());
+ final ZooUtil.LockID lock = new ZooUtil.LockID("/locks", "L1", 50);
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store = new UserFateStore<>(client, tableName, lock);
+ } else {
+ store = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock);
+ }
+
+ assertThrows(IllegalStateException.class, () -> store.reserve(fakeFateId));
+ }
+
+ @Test
+ public void testReserveReservedAndUnreserveUnreserved() throws Exception {
+ testReserveReservedAndUnreserveUnreserved(FateInstanceType.META);
+ testReserveReservedAndUnreserveUnreserved(FateInstanceType.USER);
+ }
+
+ private void testReserveReservedAndUnreserveUnreserved(FateInstanceType
storeType)
+ throws Exception {
+ final String tableName = getUniqueNames(1)[0];
+ final int numFateIds = 500;
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final List<FateStore.FateTxStore<SleepingTestEnv>> reservations = new
ArrayList<>();
+ final FateStore<SleepingTestEnv> store;
+ final ZooUtil.LockID lock = new ZooUtil.LockID("/locks", "L1", 50);
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store = new UserFateStore<>(client, tableName, lock);
+ } else {
+ store = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock);
+ }
+
+ // Create some FateIds and ensure that they can be reserved
+ for (int i = 0; i < numFateIds; i++) {
+ FateId fateId = store.create();
+ assertTrue(allIds.add(fateId));
+ var reservation = store.tryReserve(fateId);
+ assertFalse(reservation.isEmpty());
+ reservations.add(reservation.orElseThrow());
+ }
+ assertEquals(numFateIds, allIds.size());
+
+ // Try to reserve again, should not reserve
+ for (FateId fateId : allIds) {
+ assertTrue(store.tryReserve(fateId).isEmpty());
+ }
+
+ // Unreserve all the FateIds
+ for (var reservation : reservations) {
+ reservation.unreserve(Duration.ofMillis(0));
+ }
+ // Try to unreserve again (should throw exception)
+ for (var reservation : reservations) {
+ assertThrows(IllegalStateException.class, () ->
reservation.unreserve(Duration.ofMillis(0)));
+ }
+ }
+
+ @Test
+ public void testReserveAfterUnreserveAndReserveAfterDeleted() throws
Exception {
+ testReserveAfterUnreserveAndReserveAfterDeleted(FateInstanceType.META);
+ testReserveAfterUnreserveAndReserveAfterDeleted(FateInstanceType.USER);
+ }
+
+ private void
testReserveAfterUnreserveAndReserveAfterDeleted(FateInstanceType storeType)
+ throws Exception {
+ final String tableName = getUniqueNames(1)[0];
+ final int numFateIds = 500;
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final List<FateStore.FateTxStore<SleepingTestEnv>> reservations = new
ArrayList<>();
+ final FateStore<SleepingTestEnv> store;
+ final ZooUtil.LockID lock = new ZooUtil.LockID("/locks", "L1", 50);
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store = new UserFateStore<>(client, tableName, lock);
+ } else {
+ store = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock);
+ }
+
+ // Create some FateIds and ensure that they can be reserved
+ for (int i = 0; i < numFateIds; i++) {
+ FateId fateId = store.create();
+ assertTrue(allIds.add(fateId));
+ var reservation = store.tryReserve(fateId);
+ assertFalse(reservation.isEmpty());
+ reservations.add(reservation.orElseThrow());
+ }
+ assertEquals(numFateIds, allIds.size());
+
+ // Unreserve all
+ for (var reservation : reservations) {
+ reservation.unreserve(Duration.ofMillis(0));
+ }
+
+ // Ensure they can be reserved again, and delete and unreserve this time
+ for (FateId fateId : allIds) {
+ // Verify that the tx status is still NEW after unreserving since it
hasn't been deleted
+ assertEquals(ReadOnlyFateStore.TStatus.NEW,
store.read(fateId).getStatus());
+ var reservation = store.tryReserve(fateId);
+ assertFalse(reservation.isEmpty());
+ reservation.orElseThrow().delete();
+ reservation.orElseThrow().unreserve(Duration.ofMillis(0));
+ }
+
+ for (FateId fateId : allIds) {
+ // Verify that the tx is now unknown since it has been deleted
+ assertEquals(ReadOnlyFateStore.TStatus.UNKNOWN,
store.read(fateId).getStatus());
+ // Attempt to reserve a deleted txn, should throw an exception and not
wait indefinitely
+ assertThrows(IllegalStateException.class, () -> store.reserve(fateId));
+ }
+ }
+
+ @Test
+ public void testMultipleFateInstances() throws Exception {
+ testMultipleFateInstances(FateInstanceType.META);
+ testMultipleFateInstances(FateInstanceType.USER);
+ }
+
+ private void testMultipleFateInstances(FateInstanceType storeType) throws
Exception {
+ final String tableName = getUniqueNames(1)[0];
+ final int numFateIds = 500;
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final FateStore<SleepingTestEnv> store1, store2;
+ final SleepingTestEnv testEnv1 = new SleepingTestEnv(50);
+ final SleepingTestEnv testEnv2 = new SleepingTestEnv(50);
+ final ZooUtil.LockID lock1 = new ZooUtil.LockID("/locks", "L1", 50);
+ final ZooUtil.LockID lock2 = new ZooUtil.LockID("/locks", "L2", 52);
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store1 = new UserFateStore<>(client, tableName, lock1);
+ store2 = new UserFateStore<>(client, tableName, lock2);
+ } else {
+ store1 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock1);
+ store2 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock2);
+ }
+
+ Fate<SleepingTestEnv> fate1 =
+ new Fate<>(testEnv1, store1, Object::toString,
DefaultConfiguration.getInstance());
+ fate1.startDeadReservationCleaner();
+ Fate<SleepingTestEnv> fate2 =
+ new Fate<>(testEnv2, store2, Object::toString,
DefaultConfiguration.getInstance());
+ fate2.startDeadReservationCleaner();
+
+ for (int i = 0; i < numFateIds; i++) {
+ FateId fateId;
+ // Start half the txns using fate1, and the other half using fate2
+ if (i % 2 == 0) {
+ fateId = fate1.startTransaction();
+ fate1.seedTransaction("op" + i, fateId, new SleepingTestRepo(), true,
"test");
+ } else {
+ fateId = fate2.startTransaction();
+ fate2.seedTransaction("op" + i, fateId, new SleepingTestRepo(), true,
"test");
+ }
+ allIds.add(fateId);
+ }
+ assertEquals(numFateIds, allIds.size());
+
+ // Should be able to wait for completion on any fate instance
+ for (FateId fateId : allIds) {
+ fate2.waitForCompletion(fateId);
+ }
+ // Ensure that all txns have been executed and have only been executed once
+ assertTrue(Collections.disjoint(testEnv1.executedOps,
testEnv2.executedOps));
+ assertEquals(allIds, Sets.union(testEnv1.executedOps,
testEnv2.executedOps));
+
+ fate1.shutdown(1, TimeUnit.MINUTES);
+ fate2.shutdown(1, TimeUnit.MINUTES);
+ }
+
+ @Test
+ public void testDeadReservationsCleanup() throws Exception {
+ testDeadReservationsCleanup(FateInstanceType.META);
+ testDeadReservationsCleanup(FateInstanceType.USER);
+ }
+
+ private void testDeadReservationsCleanup(FateInstanceType storeType) throws
Exception {
+ // Tests reserving some transactions, then simulating that the Manager
died by creating
+ // a new Fate instance and store with a new LockID. The transactions which
were
+ // reserved using the old LockID should be cleaned up by Fate's
DeadReservationCleaner,
+ // then picked up by the new Fate/store.
+
+ final String tableName = getUniqueNames(1)[0];
+ // One transaction for each FATE worker thread
+ final int numFateIds =
+
Integer.parseInt(Property.MANAGER_FATE_THREADPOOL_SIZE.getDefaultValue());
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final FateStore<LatchTestEnv> mockedStore1, store2;
+ final LatchTestEnv testEnv1 = new LatchTestEnv();
+ final LatchTestEnv testEnv2 = new LatchTestEnv();
+ final ZooUtil.LockID lock1 = new ZooUtil.LockID("/locks", "L1", 50);
+ final ZooUtil.LockID lock2 = new ZooUtil.LockID("/locks", "L2", 52);
+ Map<FateId,FateStore.FateReservation> reservations;
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ mockedStore1 = EasyMock.createMockBuilder(UserFateStore.class)
+ .withConstructor(ClientContext.class, String.class,
ZooUtil.LockID.class)
+ .withArgs(client, tableName,
lock1).addMockedMethod("isDeadReservation").createMock();
+ } else {
+ mockedStore1 = EasyMock.createMockBuilder(MetaFateStore.class)
+ .withConstructor(String.class, ZooReaderWriter.class, ZooCache.class,
+ ZooUtil.LockID.class)
+ .withArgs(FATE_DIR, zk, client.getZooCache(),
lock1).addMockedMethod("isDeadReservation")
+ .createMock();
Review Comment:
Not sure about this, but the following pattern may simplify this test over
mocking where liveLocks is a `Set<ZooUtil.LockID>` that the test can add and
remove things to as needed.
```suggestion
mockedStore1 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(),
lock1) {
@Override
public boolean isDeadReservation(FateReservation reservation) {
return !liveLocks.contains(reservation.getLockID());
}
};
```
Another thing to consider is passing in a `Predicate<ZooUtil.LockID>` to the
Fate store constructors that can be used to check if a lock is live. Then the
`isDeadReservation()` method could be dropped an stores could use this
predicate internally. Test could pass in a predicate that checks a set. The
real code could pass in a predicate taht calls the ServiceLock code.
##########
core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java:
##########
@@ -151,6 +154,76 @@ protected void create(FateId fateId, FateKey fateKey) {
+ " and fateKey " + fateKey + " after " + maxAttempts + " attempts");
}
+ @Override
+ public Optional<FateTxStore<T>> tryReserve(FateId fateId) {
+ // Create a unique FateReservation for this reservation attempt
+ FateReservation reservation = FateReservation.from(lockID,
UUID.randomUUID());
+
+ FateMutator.Status status =
newMutator(fateId).putReservedTx(reservation).tryMutate();
+ if (status.equals(FateMutator.Status.ACCEPTED)) {
+ return Optional.of(new FateTxStoreImpl(fateId, reservation));
+ } else if (status.equals(FateMutator.Status.UNKNOWN)) {
+ // If the status is UNKNOWN, this means an error occurred after the
mutation was
+ // sent to the TabletServer, and it is unknown if the mutation was
written. We
+ // need to check if the mutation was written and if it was written by
this
+ // attempt at reservation. If it was written by this reservation attempt,
+ // we can return the FateTxStore since it was successfully reserved in
this
+ // attempt, otherwise we return empty (was written by another reservation
+ // attempt or was not written at all).
+ status = newMutator(fateId).requireReserved(reservation).tryMutate();
+ if (status.equals(FateMutator.Status.ACCEPTED)) {
+ return Optional.of(new FateTxStoreImpl(fateId, reservation));
+ }
Review Comment:
Its possible that this status could be unknown, but I don't think we need to
loop and try writing again in this case. I think we can avoid attempting
another write here altogether (because this is a tryReserve method) and just
scan and read the reservation column. If its set then we can return the
operation w/ the reservation and if its not set then we can return emptty().
So that avoids looping or retrying.
```suggestion
var persistedReservation = // todo scan and read reservation
if(reservation.equals(persistedReservation)){
return Optional.of(new FateTxStoreImpl(fateId, reservation));
} // else fall through and return empty()
```
##########
core/src/main/java/org/apache/accumulo/core/fate/AbstractFateStore.java:
##########
@@ -502,4 +520,14 @@ protected Serializable deserializeTxInfo(TxInfo txInfo,
byte[] data) {
throw new IllegalStateException("Bad node data " + txInfo);
}
}
+
+ /**
+ * TODO 4131 this is a temporary method used to create a dummy lock when
using a FateStore outside
Review Comment:
This would be a follow on issue. We can have a path in ZK for utilities to
get a ZK lock. Currently I think some of the fate admin utilities that change
the fate store require that no manager process is running. These utilities
could be changed to get a lock at this utility path in ZK and reserve fate txs
they operate on instead of requiring no manager process.
##########
core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java:
##########
@@ -151,6 +154,76 @@ protected void create(FateId fateId, FateKey fateKey) {
+ " and fateKey " + fateKey + " after " + maxAttempts + " attempts");
}
+ @Override
+ public Optional<FateTxStore<T>> tryReserve(FateId fateId) {
+ // Create a unique FateReservation for this reservation attempt
+ FateReservation reservation = FateReservation.from(lockID,
UUID.randomUUID());
+
+ FateMutator.Status status =
newMutator(fateId).putReservedTx(reservation).tryMutate();
+ if (status.equals(FateMutator.Status.ACCEPTED)) {
+ return Optional.of(new FateTxStoreImpl(fateId, reservation));
+ } else if (status.equals(FateMutator.Status.UNKNOWN)) {
+ // If the status is UNKNOWN, this means an error occurred after the
mutation was
+ // sent to the TabletServer, and it is unknown if the mutation was
written. We
+ // need to check if the mutation was written and if it was written by
this
+ // attempt at reservation. If it was written by this reservation attempt,
+ // we can return the FateTxStore since it was successfully reserved in
this
+ // attempt, otherwise we return empty (was written by another reservation
+ // attempt or was not written at all).
+ status = newMutator(fateId).requireReserved(reservation).tryMutate();
+ if (status.equals(FateMutator.Status.ACCEPTED)) {
+ return Optional.of(new FateTxStoreImpl(fateId, reservation));
+ }
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public boolean isReserved(FateId fateId) {
+ return
newMutator(fateId).requireReserved().tryMutate().equals(FateMutator.Status.ACCEPTED);
Review Comment:
This should not write a mutation to check if something is reseved, it should
scan and read the reserved column. The server side lock of a conditional
mutation does not offer benefit here because after this method returns and
something uses the returned value, the lock is no longer held. So may as well
scan in that case.
##########
core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java:
##########
@@ -151,6 +154,76 @@ protected void create(FateId fateId, FateKey fateKey) {
+ " and fateKey " + fateKey + " after " + maxAttempts + " attempts");
}
+ @Override
+ public Optional<FateTxStore<T>> tryReserve(FateId fateId) {
+ // Create a unique FateReservation for this reservation attempt
+ FateReservation reservation = FateReservation.from(lockID,
UUID.randomUUID());
+
+ FateMutator.Status status =
newMutator(fateId).putReservedTx(reservation).tryMutate();
+ if (status.equals(FateMutator.Status.ACCEPTED)) {
+ return Optional.of(new FateTxStoreImpl(fateId, reservation));
+ } else if (status.equals(FateMutator.Status.UNKNOWN)) {
+ // If the status is UNKNOWN, this means an error occurred after the
mutation was
+ // sent to the TabletServer, and it is unknown if the mutation was
written. We
+ // need to check if the mutation was written and if it was written by
this
+ // attempt at reservation. If it was written by this reservation attempt,
+ // we can return the FateTxStore since it was successfully reserved in
this
+ // attempt, otherwise we return empty (was written by another reservation
+ // attempt or was not written at all).
+ status = newMutator(fateId).requireReserved(reservation).tryMutate();
+ if (status.equals(FateMutator.Status.ACCEPTED)) {
+ return Optional.of(new FateTxStoreImpl(fateId, reservation));
+ }
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public boolean isReserved(FateId fateId) {
+ return
newMutator(fateId).requireReserved().tryMutate().equals(FateMutator.Status.ACCEPTED);
+ }
+
+ @Override
+ public Map<FateId,FateReservation> getActiveReservations() {
+ Map<FateId,FateReservation> activeReservations = new HashMap<>();
+
+ try (Scanner scanner = context.createScanner(tableName,
Authorizations.EMPTY)) {
+ scanner.setRange(new Range());
+ scanner.fetchColumn(TxColumnFamily.RESERVATION_COLUMN.getColumnFamily(),
+ TxColumnFamily.RESERVATION_COLUMN.getColumnQualifier());
+ scanner.stream()
+ .filter(entry ->
FateReservation.isFateReservation(entry.getValue().toString()))
+ .forEach(entry -> {
+ String reservationColVal = entry.getValue().toString();
+ FateId fateId = FateId.from(fateInstanceType,
entry.getKey().getRow().toString());
+ FateReservation reservation =
FateReservation.from(reservationColVal);
+ activeReservations.put(fateId, reservation);
+ });
+ } catch (TableNotFoundException e) {
+ throw new IllegalStateException(tableName + " not found!", e);
+ }
+
+ return activeReservations;
+ }
+
+ @Override
+ public void deleteDeadReservations() {
+ for (Entry<FateId,FateReservation> entry :
getActiveReservations().entrySet()) {
+ FateId fateId = entry.getKey();
+ FateReservation reservation = entry.getValue();
+ if (isDeadReservation(reservation)) {
+ newMutator(fateId).putUnreserveTx(reservation).tryMutate();
+ // No need to check the status... If it is ACCEPTED, we have
successfully unreserved
Review Comment:
Nice description
##########
core/src/main/java/org/apache/accumulo/core/fate/Fate.java:
##########
@@ -313,6 +317,26 @@ private void undo(FateId fateId, Repo<T> op) {
}
+ /**
+ * A thread that finds reservations held by dead processes and unreserves
them. Only one thread
+ * runs per store type across all Fate instances (one to clean up dead
reservations for
+ * {@link org.apache.accumulo.core.fate.user.UserFateStore UserFateStore}
and one to clean up dead
+ * reservations for {@link MetaFateStore}).
+ */
+ private class DeadReservationCleaner implements Runnable {
+ // TODO 4131 periodic check runs every 30 seconds
+ // Should this be longer? Shorter? A configurable Property? A function of
something?
+ private static final long INTERVAL_MILLIS = 30_000;
+
+ @Override
+ public void run() {
+ while (keepRunning.get()) {
+ store.deleteDeadReservations();
Review Comment:
I have not looked into it, but need to understand what happens if this
thread dies/throws an exception. Accumulo servers processes can designate some
threads as critical and that would essentially cause the manager process to die
when that is done. May want to do that for this thread, if its not already
done.
##########
core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java:
##########
@@ -272,8 +340,12 @@ public FateInstanceType type() {
private class FateTxStoreImpl extends AbstractFateTxStoreImpl<T> {
- private FateTxStoreImpl(FateId fateId, boolean isReserved) {
- super(fateId, isReserved);
+ private FateTxStoreImpl(FateId fateId) {
+ super(fateId);
+ }
+
+ private FateTxStoreImpl(FateId fateId, FateReservation reservation) {
Review Comment:
This would be a follow on issue, not something for this PR. For all of the
writes made for a Fate tx we can now add a condition requiring the reservation
to be set.
##########
test/src/main/java/org/apache/accumulo/test/fate/MultipleStoresIT.java:
##########
@@ -0,0 +1,571 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.fate;
+
+import static
org.apache.accumulo.test.fate.user.UserFateStoreIT.createFateTable;
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.conf.DefaultConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.FateId;
+import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.fate.FateStore;
+import org.apache.accumulo.core.fate.MetaFateStore;
+import org.apache.accumulo.core.fate.ReadOnlyFateStore;
+import org.apache.accumulo.core.fate.Repo;
+import org.apache.accumulo.core.fate.user.UserFateStore;
+import org.apache.accumulo.core.fate.zookeeper.ZooCache;
+import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.util.Wait;
+import org.apache.accumulo.test.zookeeper.ZooKeeperTestingServer;
+import org.easymock.EasyMock;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Sets;
+
+// TODO 4131 could potentially have separate classes for testing MetaFateStore
and UserFateStore
+// similar to how FateTestRunner is used, however that interface doesn't work
as nicely here
+// since we are using multiple stores instead of just one. Can do something
similar to
+// FateTestRunner here if desired
+public class MultipleStoresIT extends SharedMiniClusterBase {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MultipleStoresIT.class);
+ @TempDir
+ private static File tempDir;
+ private static ZooKeeperTestingServer szk = null;
+ private static ZooReaderWriter zk;
+ private static final String FATE_DIR = "/fate";
+ private ClientContext client;
+
+ @BeforeEach
+ public void beforeEachSetup() {
+ client = (ClientContext)
Accumulo.newClient().from(getClientProps()).build();
+ }
+
+ @AfterEach
+ public void afterEachTeardown() {
+ client.close();
+ }
+
+ @BeforeAll
+ public static void beforeAllSetup() throws Exception {
+ SharedMiniClusterBase.startMiniCluster();
+ szk = new ZooKeeperTestingServer(tempDir);
+ zk = szk.getZooReaderWriter();
+ }
+
+ @AfterAll
+ public static void afterAllTeardown() throws Exception {
+ SharedMiniClusterBase.stopMiniCluster();
+ szk.close();
+ }
+
+ @Test
+ public void testReserveUnreserve() throws Exception {
+ testReserveUnreserve(FateInstanceType.META);
+ testReserveUnreserve(FateInstanceType.USER);
+ }
+
+ private void testReserveUnreserve(FateInstanceType storeType) throws
Exception {
+ // reserving/unreserving a FateId should be reflected across instances of
the stores
+ final String tableName = getUniqueNames(1)[0];
+ final int numFateIds = 500;
+ final List<FateStore.FateTxStore<SleepingTestEnv>> reservations = new
ArrayList<>();
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final FateStore<SleepingTestEnv> store1, store2;
+ final ZooUtil.LockID lock1 = new ZooUtil.LockID("/locks", "L1", 50);
+ final ZooUtil.LockID lock2 = new ZooUtil.LockID("/locks", "L2", 52);
+ Map<FateId,FateStore.FateReservation> activeReservations;
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store1 = new UserFateStore<>(client, tableName, lock1);
+ store2 = new UserFateStore<>(client, tableName, lock2);
+ } else {
+ store1 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock1);
+ store2 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock2);
+ }
+
+ // Create the fate ids using store1
+ for (int i = 0; i < numFateIds; i++) {
+ assertTrue(allIds.add(store1.create()));
+ }
+ assertEquals(numFateIds, allIds.size());
+
+ // Reserve half the fate ids using store1 and rest using store2, after
reserving a fate id in
+ // one, should not be able to reserve the same in the other. Should also
not matter that all the
+ // ids were created using store1
+ int count = 0;
+ for (FateId fateId : allIds) {
+ if (count % 2 == 0) {
+ reservations.add(store1.reserve(fateId));
+ assertTrue(store2.tryReserve(fateId).isEmpty());
+ } else {
+ reservations.add(store2.reserve(fateId));
+ assertTrue(store1.tryReserve(fateId).isEmpty());
+ }
+ count++;
+ }
+ // Both stores should return the same reserved transactions
+ activeReservations = store1.getActiveReservations();
+ assertEquals(allIds, activeReservations.keySet());
+ activeReservations = store2.getActiveReservations();
+ assertEquals(allIds, activeReservations.keySet());
+
+ // Test setting/getting the TStatus and unreserving the transactions
+ for (int i = 0; i < allIds.size(); i++) {
+ var reservation = reservations.get(i);
+ assertEquals(ReadOnlyFateStore.TStatus.NEW, reservation.getStatus());
+ reservation.setStatus(ReadOnlyFateStore.TStatus.SUBMITTED);
+ assertEquals(ReadOnlyFateStore.TStatus.SUBMITTED,
reservation.getStatus());
+ reservation.delete();
+ reservation.unreserve(Duration.ofMillis(0));
+ // Attempt to set a status on a tx that has been unreserved (should
throw exception)
+ assertThrows(IllegalStateException.class,
+ () -> reservation.setStatus(ReadOnlyFateStore.TStatus.NEW));
+ }
+ assertTrue(store1.getActiveReservations().isEmpty());
+ assertTrue(store2.getActiveReservations().isEmpty());
+ }
+
+ @Test
+ public void testReserveNonExistentTxn() throws Exception {
+ testReserveNonExistentTxn(FateInstanceType.META);
+ testReserveNonExistentTxn(FateInstanceType.USER);
+ }
+
+ private void testReserveNonExistentTxn(FateInstanceType storeType) throws
Exception {
+ // Tests that reserve() doesn't hang indefinitely and instead throws an
error
+ // on reserve() a non-existent transaction.
+ final FateStore<SleepingTestEnv> store;
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final String tableName = getUniqueNames(1)[0];
+ final FateId fakeFateId = FateId.from(storeType, UUID.randomUUID());
+ final ZooUtil.LockID lock = new ZooUtil.LockID("/locks", "L1", 50);
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store = new UserFateStore<>(client, tableName, lock);
+ } else {
+ store = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock);
+ }
+
+ assertThrows(IllegalStateException.class, () -> store.reserve(fakeFateId));
+ }
+
+ @Test
+ public void testReserveReservedAndUnreserveUnreserved() throws Exception {
+ testReserveReservedAndUnreserveUnreserved(FateInstanceType.META);
+ testReserveReservedAndUnreserveUnreserved(FateInstanceType.USER);
+ }
+
+ private void testReserveReservedAndUnreserveUnreserved(FateInstanceType
storeType)
+ throws Exception {
+ final String tableName = getUniqueNames(1)[0];
+ final int numFateIds = 500;
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final List<FateStore.FateTxStore<SleepingTestEnv>> reservations = new
ArrayList<>();
+ final FateStore<SleepingTestEnv> store;
+ final ZooUtil.LockID lock = new ZooUtil.LockID("/locks", "L1", 50);
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store = new UserFateStore<>(client, tableName, lock);
+ } else {
+ store = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock);
+ }
+
+ // Create some FateIds and ensure that they can be reserved
+ for (int i = 0; i < numFateIds; i++) {
+ FateId fateId = store.create();
+ assertTrue(allIds.add(fateId));
+ var reservation = store.tryReserve(fateId);
+ assertFalse(reservation.isEmpty());
+ reservations.add(reservation.orElseThrow());
+ }
+ assertEquals(numFateIds, allIds.size());
+
+ // Try to reserve again, should not reserve
+ for (FateId fateId : allIds) {
+ assertTrue(store.tryReserve(fateId).isEmpty());
+ }
+
+ // Unreserve all the FateIds
+ for (var reservation : reservations) {
+ reservation.unreserve(Duration.ofMillis(0));
+ }
+ // Try to unreserve again (should throw exception)
+ for (var reservation : reservations) {
+ assertThrows(IllegalStateException.class, () ->
reservation.unreserve(Duration.ofMillis(0)));
+ }
+ }
+
+ @Test
+ public void testReserveAfterUnreserveAndReserveAfterDeleted() throws
Exception {
+ testReserveAfterUnreserveAndReserveAfterDeleted(FateInstanceType.META);
+ testReserveAfterUnreserveAndReserveAfterDeleted(FateInstanceType.USER);
+ }
+
+ private void
testReserveAfterUnreserveAndReserveAfterDeleted(FateInstanceType storeType)
+ throws Exception {
+ final String tableName = getUniqueNames(1)[0];
+ final int numFateIds = 500;
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final List<FateStore.FateTxStore<SleepingTestEnv>> reservations = new
ArrayList<>();
+ final FateStore<SleepingTestEnv> store;
+ final ZooUtil.LockID lock = new ZooUtil.LockID("/locks", "L1", 50);
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store = new UserFateStore<>(client, tableName, lock);
+ } else {
+ store = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock);
+ }
+
+ // Create some FateIds and ensure that they can be reserved
+ for (int i = 0; i < numFateIds; i++) {
+ FateId fateId = store.create();
+ assertTrue(allIds.add(fateId));
+ var reservation = store.tryReserve(fateId);
+ assertFalse(reservation.isEmpty());
+ reservations.add(reservation.orElseThrow());
+ }
+ assertEquals(numFateIds, allIds.size());
+
+ // Unreserve all
+ for (var reservation : reservations) {
+ reservation.unreserve(Duration.ofMillis(0));
+ }
+
+ // Ensure they can be reserved again, and delete and unreserve this time
+ for (FateId fateId : allIds) {
+ // Verify that the tx status is still NEW after unreserving since it
hasn't been deleted
+ assertEquals(ReadOnlyFateStore.TStatus.NEW,
store.read(fateId).getStatus());
+ var reservation = store.tryReserve(fateId);
+ assertFalse(reservation.isEmpty());
+ reservation.orElseThrow().delete();
+ reservation.orElseThrow().unreserve(Duration.ofMillis(0));
+ }
+
+ for (FateId fateId : allIds) {
+ // Verify that the tx is now unknown since it has been deleted
+ assertEquals(ReadOnlyFateStore.TStatus.UNKNOWN,
store.read(fateId).getStatus());
+ // Attempt to reserve a deleted txn, should throw an exception and not
wait indefinitely
+ assertThrows(IllegalStateException.class, () -> store.reserve(fateId));
+ }
+ }
+
+ @Test
+ public void testMultipleFateInstances() throws Exception {
+ testMultipleFateInstances(FateInstanceType.META);
+ testMultipleFateInstances(FateInstanceType.USER);
+ }
+
+ private void testMultipleFateInstances(FateInstanceType storeType) throws
Exception {
+ final String tableName = getUniqueNames(1)[0];
+ final int numFateIds = 500;
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final FateStore<SleepingTestEnv> store1, store2;
+ final SleepingTestEnv testEnv1 = new SleepingTestEnv(50);
+ final SleepingTestEnv testEnv2 = new SleepingTestEnv(50);
+ final ZooUtil.LockID lock1 = new ZooUtil.LockID("/locks", "L1", 50);
+ final ZooUtil.LockID lock2 = new ZooUtil.LockID("/locks", "L2", 52);
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ store1 = new UserFateStore<>(client, tableName, lock1);
+ store2 = new UserFateStore<>(client, tableName, lock2);
+ } else {
+ store1 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock1);
+ store2 = new MetaFateStore<>(FATE_DIR, zk, client.getZooCache(), lock2);
+ }
+
+ Fate<SleepingTestEnv> fate1 =
+ new Fate<>(testEnv1, store1, Object::toString,
DefaultConfiguration.getInstance());
+ fate1.startDeadReservationCleaner();
+ Fate<SleepingTestEnv> fate2 =
+ new Fate<>(testEnv2, store2, Object::toString,
DefaultConfiguration.getInstance());
+ fate2.startDeadReservationCleaner();
+
+ for (int i = 0; i < numFateIds; i++) {
+ FateId fateId;
+ // Start half the txns using fate1, and the other half using fate2
+ if (i % 2 == 0) {
+ fateId = fate1.startTransaction();
+ fate1.seedTransaction("op" + i, fateId, new SleepingTestRepo(), true,
"test");
+ } else {
+ fateId = fate2.startTransaction();
+ fate2.seedTransaction("op" + i, fateId, new SleepingTestRepo(), true,
"test");
+ }
+ allIds.add(fateId);
+ }
+ assertEquals(numFateIds, allIds.size());
+
+ // Should be able to wait for completion on any fate instance
+ for (FateId fateId : allIds) {
+ fate2.waitForCompletion(fateId);
+ }
+ // Ensure that all txns have been executed and have only been executed once
+ assertTrue(Collections.disjoint(testEnv1.executedOps,
testEnv2.executedOps));
+ assertEquals(allIds, Sets.union(testEnv1.executedOps,
testEnv2.executedOps));
+
+ fate1.shutdown(1, TimeUnit.MINUTES);
+ fate2.shutdown(1, TimeUnit.MINUTES);
+ }
+
+ @Test
+ public void testDeadReservationsCleanup() throws Exception {
+ testDeadReservationsCleanup(FateInstanceType.META);
+ testDeadReservationsCleanup(FateInstanceType.USER);
+ }
+
+ private void testDeadReservationsCleanup(FateInstanceType storeType) throws
Exception {
+ // Tests reserving some transactions, then simulating that the Manager
died by creating
+ // a new Fate instance and store with a new LockID. The transactions which
were
+ // reserved using the old LockID should be cleaned up by Fate's
DeadReservationCleaner,
+ // then picked up by the new Fate/store.
+
+ final String tableName = getUniqueNames(1)[0];
+ // One transaction for each FATE worker thread
+ final int numFateIds =
+
Integer.parseInt(Property.MANAGER_FATE_THREADPOOL_SIZE.getDefaultValue());
+ final boolean isUserStore = storeType.equals(FateInstanceType.USER);
+ final Set<FateId> allIds = new HashSet<>();
+ final FateStore<LatchTestEnv> mockedStore1, store2;
+ final LatchTestEnv testEnv1 = new LatchTestEnv();
+ final LatchTestEnv testEnv2 = new LatchTestEnv();
+ final ZooUtil.LockID lock1 = new ZooUtil.LockID("/locks", "L1", 50);
+ final ZooUtil.LockID lock2 = new ZooUtil.LockID("/locks", "L2", 52);
+ Map<FateId,FateStore.FateReservation> reservations;
+
+ if (isUserStore) {
+ createFateTable(client, tableName);
+ mockedStore1 = EasyMock.createMockBuilder(UserFateStore.class)
+ .withConstructor(ClientContext.class, String.class,
ZooUtil.LockID.class)
+ .withArgs(client, tableName,
lock1).addMockedMethod("isDeadReservation").createMock();
Review Comment:
Wonder if instead of mocking that this method could create real locks in ZK
since its running in an IT. This would make the test be a bit more end to end.
Looked a bit at ServiceLockIT trying to see what was needed to create a lock
and does not look like a few lines of code. Maybe some utility code could be
shared with that class, but not sure. If its even workable, could be a follow
on improvement. Not too sure about this suggestion, just posting for
discussion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
