Use conditional mutations for AccumuloStore. Add checks for status and putRepo

core/src/main/java/org/apache/accumulo/core/fate/accumulo/AccumuloStore.java

@@ -65,6 +65,10 @@ public AccumuloStore(ClientContext context, String tableName) {
   public long create() {
     long tid = RANDOM.get().nextLong() & 0x7fffffffffffffffL;
 
+    // once requireAbsentTransaction() is implemented, use it here
+    // newMutator(tid).requireAbsentTransaction().putStatus(TStatus.NEW)
+    // .putCreateTime(System.currentTimeMillis()).mutate();
+
     newMutator(tid).putStatus(TStatus.NEW).putCreateTime(System.currentTimeMillis()).mutate();
 
     return tid;
@@ -236,27 +240,9 @@ public void setStatus(TStatus status) {
     public void setTransactionInfo(TxInfo txInfo, Serializable so) {
       verifyReserved(true);
 
-      FateMutator<T> fateMutator = newMutator(tid);
       final byte[] serialized = serializeTxInfo(so);
 
-      switch (txInfo) {
-        case TX_NAME:
-          fateMutator.putName(serialized);
-          break;
-        case AUTO_CLEAN:
-          fateMutator.putAutoClean(serialized);
-          break;
-        case EXCEPTION:
-          fateMutator.putException(serialized);
-          break;
-        case RETURN_VALUE:
-          fateMutator.putReturnValue(serialized);
-          break;
-        default:
-          throw new IllegalArgumentException("Unexpected TxInfo type " + txInfo);
-      }
-
-      fateMutator.mutate();
+      newMutator(tid).putTxInfo(txInfo, serialized).mutate();
     }
 
     @Override

core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutatorImpl.java

@@ -24,11 +24,14 @@
 
 import java.util.Objects;
 
-import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.ConditionalWriter;
 import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.clientImpl.ClientContext;
-import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Condition;
+import org.apache.accumulo.core.data.ConditionalMutation;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.fate.Fate.TxInfo;
 import org.apache.accumulo.core.fate.ReadOnlyFateStore.TStatus;
@@ -45,13 +48,13 @@ public class FateMutatorImpl<T> implements FateMutator<T> {
   private final ClientContext context;
   private final String tableName;
   private final long tid;
-  private final Mutation mutation;
+  private final ConditionalMutation mutation;
 
-  FateMutatorImpl(ClientContext context, String tableName, long tid) {
+  public FateMutatorImpl(ClientContext context, String tableName, long tid) {
     this.context = Objects.requireNonNull(context);
     this.tableName = Objects.requireNonNull(tableName);
     this.tid = tid;
-    this.mutation = new Mutation(new Text("tx_" + FastFormat.toHexString(tid)));
+    this.mutation = new ConditionalMutation(new Text("tx_" + FastFormat.toHexString(tid)));
   }
 
   @Override
@@ -105,13 +108,20 @@ public FateMutator<T> putTxInfo(TxInfo txInfo, byte[] data) {
       case RETURN_VALUE:
         putReturnValue(data);
         break;
+      default:
+        throw new IllegalArgumentException("Unexpected TxInfo type " + txInfo);
     }
     return this;
   }
 
   @Override
   public FateMutator<T> putRepo(int position, Repo<T> repo) {
-    mutation.put(RepoColumnFamily.NAME, invertRepo(position), new Value(serialize(repo)));
+    final Text cq = invertRepo(position);
+    // ensure this repo is not already set
+    mutation.addCondition(new Condition(RepoColumnFamily.NAME, cq));
+    // TODO: would be nice to make sure that the previous repo is there but not sure we can do that
+    // without knowing its value
+    mutation.put(RepoColumnFamily.NAME, cq, new Value(serialize(repo)));
     return this;
   }
 
@@ -132,12 +142,29 @@ public FateMutator<T> delete() {
     return this;
   }
 
+  /**
+   * Require that the transaction does not exist.
+   */
+  // TODO: need to figure out how to use TabletExistsIterator since its in a different module
+  // public FateMutator<T> requireAbsentTransaction() {
+  // IteratorSetting is = new IteratorSetting(1000000, TabletExistsIterator.class);
+  // Condition c = new Condition("", "").setIterators(is);
+  // mutation.addCondition(c);
+  // return this;
+  // }
+
   @Override
   public void mutate() {
-    try (BatchWriter writer = context.createBatchWriter(tableName)) {
-      writer.addMutation(mutation);
-    } catch (Exception e) {
-      throw new IllegalStateException(e);
+    try (ConditionalWriter writer = context.createConditionalWriter(tableName)) {
+      if (mutation.getConditions().isEmpty()) {
+        mutation.addCondition(new Condition("", ""));
+      }
+      ConditionalWriter.Result result = writer.write(mutation);
+      if (result.getStatus() != ConditionalWriter.Status.ACCEPTED) {
+        throw new IllegalStateException("Failed to write mutation " + mutation);
+      }
+    } catch (AccumuloException | TableNotFoundException | AccumuloSecurityException e) {
+      throw new RuntimeException(e);
     }
   }
 }

test/src/main/java/org/apache/accumulo/test/fate/accumulo/FateMutatorImplIT.java

@@ -0,0 +1,94 @@
+/*
+ * 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.accumulo;
+
+import static org.apache.accumulo.core.util.LazySingletons.RANDOM;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.time.Duration;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.client.admin.TabletHostingGoal;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.fate.accumulo.FateMutatorImpl;
+import org.apache.accumulo.harness.SharedMiniClusterBase;
+import org.apache.accumulo.test.fate.FateIT;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FateMutatorImplIT extends SharedMiniClusterBase {
+
+  Logger log = LoggerFactory.getLogger(FateMutatorImplIT.class);
+  final NewTableConfiguration ntc =
+      new NewTableConfiguration().withInitialHostingGoal(TabletHostingGoal.ALWAYS);
+
+  @BeforeAll
+  public static void setup() throws Exception {
+    SharedMiniClusterBase.startMiniCluster();
+  }
+
+  @AfterAll
+  public static void tearDown() {
+    SharedMiniClusterBase.stopMiniCluster();
+  }
+
+  @Override
+  protected Duration defaultTimeout() {
+    return Duration.ofMinutes(5);
+  }
+
+  @Test
+  public void putRepo() throws Exception {
+    final String table = getUniqueNames(1)[0];
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+      client.tableOperations().create(table, ntc);
+
+      ClientContext context = (ClientContext) client;
+
+      final long tid = RANDOM.get().nextLong() & 0x7fffffffffffffffL;
+
+      // add some repos in order
+      FateMutatorImpl<FateIT.TestEnv> fateMutator = new FateMutatorImpl<>(context, table, tid);
+      fateMutator.putRepo(100, new FateIT.TestRepo("test")).mutate();
+      FateMutatorImpl<FateIT.TestEnv> fateMutator1 = new FateMutatorImpl<>(context, table, tid);
+      fateMutator1.putRepo(99, new FateIT.TestRepo("test")).mutate();
+      FateMutatorImpl<FateIT.TestEnv> fateMutator2 = new FateMutatorImpl<>(context, table, tid);
+      fateMutator2.putRepo(98, new FateIT.TestRepo("test")).mutate();
+
+      // make sure we cant add a repo that has already been added
+      FateMutatorImpl<FateIT.TestEnv> fateMutator3 = new FateMutatorImpl<>(context, table, tid);
+      assertThrows(IllegalStateException.class,
+          () -> fateMutator3.putRepo(98, new FateIT.TestRepo("test")).mutate(),
+          "Repo in position 98 already exists. Expected to not be able to add it again.");
+      FateMutatorImpl<FateIT.TestEnv> fateMutator4 = new FateMutatorImpl<>(context, table, tid);
+      assertThrows(IllegalStateException.class,
+          () -> fateMutator4.putRepo(99, new FateIT.TestRepo("test")).mutate(),
+          "Repo in position 99 already exists. Expected to not be able to add it again.");
+    }
+  }
+
+  void logAllEntriesInTable(String tableName, AccumuloClient client) throws Exception {
+    client.createScanner(tableName).forEach(e -> log.info(e.getKey() + " " + e.getValue()));
+  }
+}