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

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

@@ -70,7 +70,8 @@ public AccumuloStore(ClientContext context) {
   public long create() {
     long tid = RANDOM.get().nextLong() & 0x7fffffffffffffffL;
 
-    newMutator(tid).putStatus(TStatus.NEW).putCreateTime(System.currentTimeMillis()).mutate();
+    newMutator(tid).requireStatus().putStatus(TStatus.NEW).putCreateTime(System.currentTimeMillis())
+        .mutate();
 
     return tid;
   }
@@ -241,27 +242,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,18 @@ 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));
+    mutation.put(RepoColumnFamily.NAME, cq, new Value(serialize(repo)));
     return this;
   }
 
@@ -132,12 +140,30 @@ public FateMutator<T> delete() {
     return this;
   }
 
+  /**
+   * Require that the transaction status is one of the given statuses. If no statuses are provided,
+   * require that the status column is absent.
+   *
+   * @param statuses The statuses to check against.
+   */
+  public FateMutator<T> requireStatus(TStatus... statuses) {
+    Condition condition = StatusMappingIterator.createCondition(statuses);
+    mutation.addCondition(condition);
+    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);
     }
   }
 }

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

@@ -0,0 +1,148 @@
+/*
+ * 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.core.fate.accumulo;
+
+import static org.apache.accumulo.core.fate.accumulo.schema.FateSchema.TxColumnFamily.STATUS_COLUMN;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Condition;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.fate.ReadOnlyFateStore;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+
+/**
+ * A specialized iterator that maps the value of the status column to "present" or "absent". This
+ * iterator allows for checking of the status column's value against a set of acceptable statuses
+ * within a conditional mutation.
+ */
+public class StatusMappingIterator implements SortedKeyValueIterator<Key,Value> {
+
+  private static final String PRESENT = "present";
+  private static final String ABSENT = "absent";
+  private static final String STATUS_SET_KEY = "statusSet";
+
+  private SortedKeyValueIterator<Key,Value> source;
+  private final Set<String> acceptableStatuses = new HashSet<>();
+  private Value mappedValue;
+
+  /**
+   * The set of acceptable must be provided as an option to the iterator using the
+   * {@link #STATUS_SET_KEY} key.
+   */
+  @Override
+  public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options,
+      IteratorEnvironment env) throws IOException {
+    this.source = source;
+    if (options.containsKey(STATUS_SET_KEY)) {
+      String[] statuses = decodeStatuses(options.get(STATUS_SET_KEY));
+      acceptableStatuses.addAll(Arrays.asList(statuses));
+    }
+  }
+
+  @Override
+  public boolean hasTop() {
+    return source.hasTop();
+  }
+
+  @Override
+  public void next() throws IOException {
+    source.next();
+    mapValue();
+  }
+
+  @Override
+  public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive)
+      throws IOException {
+    source.seek(range, columnFamilies, inclusive);
+    mapValue();
+  }
+
+  /**
+   * Maps the value of the status column to "present" or "absent" based on its presence within the
+   * set of statuses.
+   */
+  private void mapValue() {
+    if (source.hasTop()) {
+      String currentValue = source.getTopValue().toString();
+      mappedValue =
+          acceptableStatuses.contains(currentValue) ? new Value(PRESENT) : new Value(ABSENT);
+    }
+  }
+
+  @Override
+  public Key getTopKey() {
+    return source.getTopKey();
+  }
+
+  @Override
+  public Value getTopValue() {
+    return mappedValue;
+  }
+
+  @Override
+  public SortedKeyValueIterator<Key,Value> deepCopy(IteratorEnvironment env) {
+    throw new UnsupportedOperationException();
+  }
+
+  /**
+   * Creates a condition that checks if the status column's value is one of the given acceptable
+   * statuses.
+   *
+   * @param statuses The acceptable statuses.
+   * @return A condition configured to use this iterator.
+   */
+  public static Condition createCondition(ReadOnlyFateStore.TStatus... statuses) {
+    Condition condition =
+        new Condition(STATUS_COLUMN.getColumnFamily(), STATUS_COLUMN.getColumnQualifier());
+
+    if (statuses.length == 0) {
+      // If no statuses are provided, require the status column to be absent. Return the condition
+      // with no value set so that the mutation will be rejected if the status column is present.
+      return condition;
+    } else {
+      IteratorSetting is = new IteratorSetting(100, StatusMappingIterator.class);
+      is.addOption(STATUS_SET_KEY, encodeStatuses(statuses));
+
+      // If the value of the status column is in the set, it will be mapped to "present", so set the
+      // value of the condition to "present".
+      return condition.setValue(PRESENT).setIterators(is);
+    }
+  }
+
+  private static String encodeStatuses(ReadOnlyFateStore.TStatus[] statuses) {
+    return Arrays.stream(statuses).map(Enum::name).collect(Collectors.joining(","));
+  }
+
+  private static String[] decodeStatuses(String statuses) {
+    return statuses.split(",");
+  }
+
+}