/*
 * 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
 *
 *   http://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;

import static org.apache.accumulo.fate.util.UtilWaitThread.sleepUninterruptibly;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.accumulo.cluster.ClusterUser;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.ConditionalWriter;
import org.apache.accumulo.core.client.ConditionalWriter.Result;
import org.apache.accumulo.core.client.ConditionalWriter.Status;
import org.apache.accumulo.core.client.ConditionalWriterConfig;
import org.apache.accumulo.core.client.IsolatedScanner;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.RowIterator;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableDeletedException;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.TableOfflineException;
import org.apache.accumulo.core.client.admin.NewTableConfiguration;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.ArrayByteSequence;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Condition;
import org.apache.accumulo.core.data.ConditionalMutation;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.LongCombiner.Type;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.WrappingIterator;
import org.apache.accumulo.core.iterators.user.SummingCombiner;
import org.apache.accumulo.core.iterators.user.VersioningIterator;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.security.SystemPermission;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.core.util.FastFormat;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
import org.apache.accumulo.test.constraints.AlphaNumKeyConstraint;
import org.apache.accumulo.test.functional.BadIterator;
import org.apache.accumulo.test.functional.SlowIterator;
import org.apache.accumulo.tracer.TraceDump;
import org.apache.accumulo.tracer.TraceServer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.htrace.Sampler;
import org.apache.htrace.Trace;
import org.apache.htrace.TraceScope;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Iterables;

public class ConditionalWriterIT extends SharedMiniClusterBase {
  private static final Logger log = LoggerFactory.getLogger(ConditionalWriterIT.class);

  @Override
  protected int defaultTimeoutSeconds() {
    return 120;
  }

  @BeforeClass
  public static void setup() throws Exception {
    SharedMiniClusterBase.startMiniClusterWithConfig(new Callback());
  }

  @AfterClass
  public static void teardown() {
    SharedMiniClusterBase.stopMiniCluster();
  }

  private static class Callback implements MiniClusterConfigurationCallback {
    @Override
    public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration coreSite) {
      // Set the min span to 0 so we will definitely get all the traces back. See ACCUMULO-4365
      Map<String,String> siteConf = cfg.getSiteConfig();
      siteConf.put(Property.TRACE_SPAN_RECEIVER_PREFIX.getKey() + "tracer.span.min.ms", "0");
      cfg.setSiteConfig(siteConf);
    }
  }

  public static long abs(long l) {
    l = Math.abs(l); // abs(Long.MIN_VALUE) == Long.MIN_VALUE...
    if (l < 0)
      return 0;
    return l;
  }

  @Before
  public void deleteUsers() throws Exception {
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      Set<String> users = client.securityOperations().listLocalUsers();
      ClusterUser user = getUser(0);
      if (users.contains(user.getPrincipal())) {
        client.securityOperations().dropLocalUser(user.getPrincipal());
      }
    }
  }

  @Test
  public void testBasic() throws Exception {

    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      client.tableOperations().create(tableName);

      try (
          ConditionalWriter cw =
              client.createConditionalWriter(tableName, new ConditionalWriterConfig());
          Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {

        // mutation conditional on column tx:seq not existing
        ConditionalMutation cm0 = new ConditionalMutation("99006", new Condition("tx", "seq"));
        cm0.put("name", "last", "doe");
        cm0.put("name", "first", "john");
        cm0.put("tx", "seq", "1");
        assertEquals(Status.ACCEPTED, cw.write(cm0).getStatus());
        assertEquals(Status.REJECTED, cw.write(cm0).getStatus());

        // mutation conditional on column tx:seq being 1
        ConditionalMutation cm1 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setValue("1"));
        cm1.put("name", "last", "Doe");
        cm1.put("tx", "seq", "2");
        assertEquals(Status.ACCEPTED, cw.write(cm1).getStatus());

        // test condition where value differs
        ConditionalMutation cm2 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setValue("1"));
        cm2.put("name", "last", "DOE");
        cm2.put("tx", "seq", "2");
        assertEquals(Status.REJECTED, cw.write(cm2).getStatus());

        // test condition where column does not exists
        ConditionalMutation cm3 =
            new ConditionalMutation("99006", new Condition("txtypo", "seq").setValue("1"));
        cm3.put("name", "last", "deo");
        cm3.put("tx", "seq", "2");
        assertEquals(Status.REJECTED, cw.write(cm3).getStatus());

        // test two conditions, where one should fail
        ConditionalMutation cm4 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setValue("2"),
                new Condition("name", "last").setValue("doe"));
        cm4.put("name", "last", "deo");
        cm4.put("tx", "seq", "3");
        assertEquals(Status.REJECTED, cw.write(cm4).getStatus());

        // test two conditions, where one should fail
        ConditionalMutation cm5 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setValue("1"),
                new Condition("name", "last").setValue("Doe"));
        cm5.put("name", "last", "deo");
        cm5.put("tx", "seq", "3");
        assertEquals(Status.REJECTED, cw.write(cm5).getStatus());

        // ensure rejected mutations did not write
        scanner.fetchColumn("name", "last");
        scanner.setRange(new Range("99006"));
        Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
        assertEquals("Doe", entry.getValue().toString());

        // test w/ two conditions that are met
        ConditionalMutation cm6 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setValue("2"),
                new Condition("name", "last").setValue("Doe"));
        cm6.put("name", "last", "DOE");
        cm6.put("tx", "seq", "3");
        assertEquals(Status.ACCEPTED, cw.write(cm6).getStatus());

        entry = Iterables.getOnlyElement(scanner);
        assertEquals("DOE", entry.getValue().toString());

        // test a conditional mutation that deletes
        ConditionalMutation cm7 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setValue("3"));
        cm7.putDelete("name", "last");
        cm7.putDelete("name", "first");
        cm7.putDelete("tx", "seq");
        assertEquals(Status.ACCEPTED, cw.write(cm7).getStatus());

        assertFalse("Did not expect to find any results", scanner.iterator().hasNext());

        // add the row back
        assertEquals(Status.ACCEPTED, cw.write(cm0).getStatus());
        assertEquals(Status.REJECTED, cw.write(cm0).getStatus());

        entry = Iterables.getOnlyElement(scanner);
        assertEquals("doe", entry.getValue().toString());
      }
    }
  }

  @Test
  public void testFields() throws Exception {

    try (AccumuloClient client1 = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      String user = null;

      ClusterUser user1 = getUser(0);
      user = user1.getPrincipal();
      if (saslEnabled()) {
        // The token is pointless for kerberos
        client1.securityOperations().createLocalUser(user, null);
      } else {
        client1.securityOperations().createLocalUser(user, new PasswordToken(user1.getPassword()));
      }

      Authorizations auths = new Authorizations("A", "B");

      client1.securityOperations().changeUserAuthorizations(user, auths);
      client1.securityOperations().grantSystemPermission(user, SystemPermission.CREATE_TABLE);

      try (AccumuloClient client2 =
          Accumulo.newClient().from(client1.properties()).as(user, user1.getToken()).build()) {
        client2.tableOperations().create(tableName);

        try (
            ConditionalWriter cw = client2.createConditionalWriter(tableName,
                new ConditionalWriterConfig().setAuthorizations(auths));
            Scanner scanner = client2.createScanner(tableName, auths)) {

          ColumnVisibility cva = new ColumnVisibility("A");
          ColumnVisibility cvb = new ColumnVisibility("B");

          ConditionalMutation cm0 =
              new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cva));
          cm0.put("name", "last", cva, "doe");
          cm0.put("name", "first", cva, "john");
          cm0.put("tx", "seq", cva, "1");
          assertEquals(Status.ACCEPTED, cw.write(cm0).getStatus());

          scanner.setRange(new Range("99006"));
          // TODO verify all columns
          scanner.fetchColumn("tx", "seq");
          Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
          assertEquals("1", entry.getValue().toString());
          long ts = entry.getKey().getTimestamp();

          // test wrong colf
          ConditionalMutation cm1 = new ConditionalMutation("99006",
              new Condition("txA", "seq").setVisibility(cva).setValue("1"));
          cm1.put("name", "last", cva, "Doe");
          cm1.put("name", "first", cva, "John");
          cm1.put("tx", "seq", cva, "2");
          assertEquals(Status.REJECTED, cw.write(cm1).getStatus());

          // test wrong colq
          ConditionalMutation cm2 = new ConditionalMutation("99006",
              new Condition("tx", "seqA").setVisibility(cva).setValue("1"));
          cm2.put("name", "last", cva, "Doe");
          cm2.put("name", "first", cva, "John");
          cm2.put("tx", "seq", cva, "2");
          assertEquals(Status.REJECTED, cw.write(cm2).getStatus());

          // test wrong colv
          ConditionalMutation cm3 = new ConditionalMutation("99006",
              new Condition("tx", "seq").setVisibility(cvb).setValue("1"));
          cm3.put("name", "last", cva, "Doe");
          cm3.put("name", "first", cva, "John");
          cm3.put("tx", "seq", cva, "2");
          assertEquals(Status.REJECTED, cw.write(cm3).getStatus());

          // test wrong timestamp
          ConditionalMutation cm4 = new ConditionalMutation("99006",
              new Condition("tx", "seq").setVisibility(cva).setTimestamp(ts + 1).setValue("1"));
          cm4.put("name", "last", cva, "Doe");
          cm4.put("name", "first", cva, "John");
          cm4.put("tx", "seq", cva, "2");
          assertEquals(Status.REJECTED, cw.write(cm4).getStatus());

          // test wrong timestamp
          ConditionalMutation cm5 = new ConditionalMutation("99006",
              new Condition("tx", "seq").setVisibility(cva).setTimestamp(ts - 1).setValue("1"));
          cm5.put("name", "last", cva, "Doe");
          cm5.put("name", "first", cva, "John");
          cm5.put("tx", "seq", cva, "2");
          assertEquals(Status.REJECTED, cw.write(cm5).getStatus());

          // ensure no updates were made
          entry = Iterables.getOnlyElement(scanner);
          assertEquals("1", entry.getValue().toString());

          // set all columns correctly
          ConditionalMutation cm6 = new ConditionalMutation("99006",
              new Condition("tx", "seq").setVisibility(cva).setTimestamp(ts).setValue("1"));
          cm6.put("name", "last", cva, "Doe");
          cm6.put("name", "first", cva, "John");
          cm6.put("tx", "seq", cva, "2");
          assertEquals(Status.ACCEPTED, cw.write(cm6).getStatus());

          entry = Iterables.getOnlyElement(scanner);
          assertEquals("2", entry.getValue().toString());
        }
      }
    }
  }

  @Test
  public void testBadColVis() throws Exception {
    // test when a user sets a col vis in a condition that can never be seen

    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      client.tableOperations().create(tableName);

      Authorizations auths = new Authorizations("A", "B");

      client.securityOperations().changeUserAuthorizations(getAdminPrincipal(), auths);

      Authorizations filteredAuths = new Authorizations("A");

      ColumnVisibility cva = new ColumnVisibility("A");
      ColumnVisibility cvb = new ColumnVisibility("B");
      ColumnVisibility cvc = new ColumnVisibility("C");

      try (ConditionalWriter cw = client.createConditionalWriter(tableName,
          new ConditionalWriterConfig().setAuthorizations(filteredAuths))) {

        // User has authorization, but didn't include it in the writer
        ConditionalMutation cm0 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb));
        cm0.put("name", "last", cva, "doe");
        cm0.put("name", "first", cva, "john");
        cm0.put("tx", "seq", cva, "1");
        assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm0).getStatus());

        ConditionalMutation cm1 = new ConditionalMutation("99006",
            new Condition("tx", "seq").setVisibility(cvb).setValue("1"));
        cm1.put("name", "last", cva, "doe");
        cm1.put("name", "first", cva, "john");
        cm1.put("tx", "seq", cva, "1");
        assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm1).getStatus());

        // User does not have the authorization
        ConditionalMutation cm2 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvc));
        cm2.put("name", "last", cva, "doe");
        cm2.put("name", "first", cva, "john");
        cm2.put("tx", "seq", cva, "1");
        assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm2).getStatus());

        ConditionalMutation cm3 = new ConditionalMutation("99006",
            new Condition("tx", "seq").setVisibility(cvc).setValue("1"));
        cm3.put("name", "last", cva, "doe");
        cm3.put("name", "first", cva, "john");
        cm3.put("tx", "seq", cva, "1");
        assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm3).getStatus());

        // if any visibility is bad, good visibilities don't override
        ConditionalMutation cm4 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb),
                new Condition("tx", "seq").setVisibility(cva));

        cm4.put("name", "last", cva, "doe");
        cm4.put("name", "first", cva, "john");
        cm4.put("tx", "seq", cva, "1");
        assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm4).getStatus());

        ConditionalMutation cm5 = new ConditionalMutation("99006",
            new Condition("tx", "seq").setVisibility(cvb).setValue("1"),
            new Condition("tx", "seq").setVisibility(cva).setValue("1"));
        cm5.put("name", "last", cva, "doe");
        cm5.put("name", "first", cva, "john");
        cm5.put("tx", "seq", cva, "1");
        assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm5).getStatus());

        ConditionalMutation cm6 = new ConditionalMutation("99006",
            new Condition("tx", "seq").setVisibility(cvb).setValue("1"),
            new Condition("tx", "seq").setVisibility(cva));
        cm6.put("name", "last", cva, "doe");
        cm6.put("name", "first", cva, "john");
        cm6.put("tx", "seq", cva, "1");
        assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm6).getStatus());

        ConditionalMutation cm7 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb),
                new Condition("tx", "seq").setVisibility(cva).setValue("1"));
        cm7.put("name", "last", cva, "doe");
        cm7.put("name", "first", cva, "john");
        cm7.put("tx", "seq", cva, "1");
        assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm7).getStatus());

      }

      // test passing auths that exceed users configured auths

      Authorizations exceedingAuths = new Authorizations("A", "B", "D");
      try (ConditionalWriter cw2 = client.createConditionalWriter(tableName,
          new ConditionalWriterConfig().setAuthorizations(exceedingAuths))) {

        ConditionalMutation cm8 =
            new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb),
                new Condition("tx", "seq").setVisibility(cva).setValue("1"));
        cm8.put("name", "last", cva, "doe");
        cm8.put("name", "first", cva, "john");
        cm8.put("tx", "seq", cva, "1");

        try {
          Status status = cw2.write(cm8).getStatus();
          fail(
              "Writing mutation with Authorizations the user doesn't have should fail. Got status: "
                  + status);
        } catch (AccumuloSecurityException ase) {
          // expected, check specific failure?
        }
      }
    }
  }

  @Test
  public void testConstraints() throws Exception {
    // ensure constraint violations are properly reported

    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      client.tableOperations().create(tableName);
      client.tableOperations().addConstraint(tableName, AlphaNumKeyConstraint.class.getName());
      client.tableOperations().clone(tableName, tableName + "_clone", true, new HashMap<>(),
          new HashSet<>());

      try (
          ConditionalWriter cw =
              client.createConditionalWriter(tableName + "_clone", new ConditionalWriterConfig());
          Scanner scanner = client.createScanner(tableName + "_clone", new Authorizations())) {

        ConditionalMutation cm0 = new ConditionalMutation("99006+", new Condition("tx", "seq"));
        cm0.put("tx", "seq", "1");

        assertEquals(Status.VIOLATED, cw.write(cm0).getStatus());
        assertFalse("Should find no results in the table is mutation result was violated",
            scanner.iterator().hasNext());

        ConditionalMutation cm1 = new ConditionalMutation("99006", new Condition("tx", "seq"));
        cm1.put("tx", "seq", "1");

        assertEquals(Status.ACCEPTED, cw.write(cm1).getStatus());
        assertTrue("Accepted result should be returned when reading table",
            scanner.iterator().hasNext());
      }
    }
  }

  @Test
  public void testIterators() throws Exception {

    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      client.tableOperations().create(tableName,
          new NewTableConfiguration().withoutDefaultIterators());

      try (BatchWriter bw = client.createBatchWriter(tableName)) {

        Mutation m = new Mutation("ACCUMULO-1000");
        m.put("count", "comments", "1");
        bw.addMutation(m);
        bw.addMutation(m);
        bw.addMutation(m);

        m = new Mutation("ACCUMULO-1001");
        m.put("count2", "comments", "1");
        bw.addMutation(m);
        bw.addMutation(m);

        m = new Mutation("ACCUMULO-1002");
        m.put("count2", "comments", "1");
        bw.addMutation(m);
        bw.addMutation(m);
      }

      IteratorSetting iterConfig = new IteratorSetting(10, SummingCombiner.class);
      SummingCombiner.setEncodingType(iterConfig, Type.STRING);
      SummingCombiner.setColumns(iterConfig,
          Collections.singletonList(new IteratorSetting.Column("count")));

      IteratorSetting iterConfig2 = new IteratorSetting(10, SummingCombiner.class);
      SummingCombiner.setEncodingType(iterConfig2, Type.STRING);
      SummingCombiner.setColumns(iterConfig2,
          Collections.singletonList(new IteratorSetting.Column("count2", "comments")));

      IteratorSetting iterConfig3 = new IteratorSetting(5, VersioningIterator.class);
      VersioningIterator.setMaxVersions(iterConfig3, 1);

      try (Scanner scanner = client.createScanner(tableName, new Authorizations())) {
        scanner.addScanIterator(iterConfig);
        scanner.setRange(new Range("ACCUMULO-1000"));
        scanner.fetchColumn("count", "comments");

        Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
        assertEquals("3", entry.getValue().toString());

        try (ConditionalWriter cw =
            client.createConditionalWriter(tableName, new ConditionalWriterConfig())) {

          ConditionalMutation cm0 = new ConditionalMutation("ACCUMULO-1000",
              new Condition("count", "comments").setValue("3"));
          cm0.put("count", "comments", "1");
          assertEquals(Status.REJECTED, cw.write(cm0).getStatus());
          entry = Iterables.getOnlyElement(scanner);
          assertEquals("3", entry.getValue().toString());

          ConditionalMutation cm1 = new ConditionalMutation("ACCUMULO-1000",
              new Condition("count", "comments").setIterators(iterConfig).setValue("3"));
          cm1.put("count", "comments", "1");
          assertEquals(Status.ACCEPTED, cw.write(cm1).getStatus());
          entry = Iterables.getOnlyElement(scanner);
          assertEquals("4", entry.getValue().toString());

          ConditionalMutation cm2 = new ConditionalMutation("ACCUMULO-1000",
              new Condition("count", "comments").setValue("4"));
          cm2.put("count", "comments", "1");
          assertEquals(Status.REJECTED, cw.write(cm1).getStatus());
          entry = Iterables.getOnlyElement(scanner);
          assertEquals("4", entry.getValue().toString());

          // run test with multiple iterators passed in same batch and condition with two iterators

          ConditionalMutation cm3 = new ConditionalMutation("ACCUMULO-1000",
              new Condition("count", "comments").setIterators(iterConfig).setValue("4"));
          cm3.put("count", "comments", "1");

          ConditionalMutation cm4 = new ConditionalMutation("ACCUMULO-1001",
              new Condition("count2", "comments").setIterators(iterConfig2).setValue("2"));
          cm4.put("count2", "comments", "1");

          ConditionalMutation cm5 =
              new ConditionalMutation("ACCUMULO-1002", new Condition("count2", "comments")
                  .setIterators(iterConfig2, iterConfig3).setValue("2"));
          cm5.put("count2", "comments", "1");

          Iterator<Result> results = cw.write(Arrays.asList(cm3, cm4, cm5).iterator());
          Map<String,Status> actual = new HashMap<>();

          while (results.hasNext()) {
            Result result = results.next();
            String k = new String(result.getMutation().getRow());
            assertFalse("Did not expect to see multiple resultus for the row: " + k,
                actual.containsKey(k));
            actual.put(k, result.getStatus());
          }

          Map<String,Status> expected = new HashMap<>();
          expected.put("ACCUMULO-1000", Status.ACCEPTED);
          expected.put("ACCUMULO-1001", Status.ACCEPTED);
          expected.put("ACCUMULO-1002", Status.REJECTED);

          assertEquals(expected, actual);
        }
      }
    }
  }

  public static class AddingIterator extends WrappingIterator {
    long amount = 0;

    @Override
    public Value getTopValue() {
      Value val = super.getTopValue();
      long l = Long.parseLong(val.toString());
      String newVal = (l + amount) + "";
      return new Value(newVal);
    }

    @Override
    public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options,
        IteratorEnvironment env) {
      this.setSource(source);
      amount = Long.parseLong(options.get("amount"));
    }
  }

  public static class MultiplyingIterator extends WrappingIterator {
    long amount = 0;

    @Override
    public Value getTopValue() {
      Value val = super.getTopValue();
      long l = Long.parseLong(val.toString());
      String newVal = l * amount + "";
      return new Value(newVal);
    }

    @Override
    public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options,
        IteratorEnvironment env) {
      this.setSource(source);
      amount = Long.parseLong(options.get("amount"));
    }
  }

  @Test
  public void testTableAndConditionIterators() throws Exception {

    // test w/ table that has iterators configured
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      IteratorSetting aiConfig1 = new IteratorSetting(30, "AI1", AddingIterator.class);
      aiConfig1.addOption("amount", "2");
      IteratorSetting aiConfig2 = new IteratorSetting(35, "MI1", MultiplyingIterator.class);
      aiConfig2.addOption("amount", "3");
      IteratorSetting aiConfig3 = new IteratorSetting(40, "AI2", AddingIterator.class);
      aiConfig3.addOption("amount", "5");

      client.tableOperations().create(tableName);

      try (BatchWriter bw = client.createBatchWriter(tableName)) {
        Mutation m = new Mutation("ACCUMULO-1000");
        m.put("count", "comments", "6");
        bw.addMutation(m);

        m = new Mutation("ACCUMULO-1001");
        m.put("count", "comments", "7");
        bw.addMutation(m);

        m = new Mutation("ACCUMULO-1002");
        m.put("count", "comments", "8");
        bw.addMutation(m);
      }

      client.tableOperations().attachIterator(tableName, aiConfig1, EnumSet.of(IteratorScope.scan));
      client.tableOperations().offline(tableName, true);
      client.tableOperations().online(tableName, true);

      try (
          ConditionalWriter cw =
              client.createConditionalWriter(tableName, new ConditionalWriterConfig());
          Scanner scanner = client.createScanner(tableName, new Authorizations())) {

        ConditionalMutation cm6 = new ConditionalMutation("ACCUMULO-1000",
            new Condition("count", "comments").setValue("8"));
        cm6.put("count", "comments", "7");
        assertEquals(Status.ACCEPTED, cw.write(cm6).getStatus());

        scanner.setRange(new Range("ACCUMULO-1000"));
        scanner.fetchColumn("count", "comments");

        Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
        assertEquals("9", entry.getValue().toString());

        ConditionalMutation cm7 = new ConditionalMutation("ACCUMULO-1000",
            new Condition("count", "comments").setIterators(aiConfig2).setValue("27"));
        cm7.put("count", "comments", "8");
        assertEquals(Status.ACCEPTED, cw.write(cm7).getStatus());

        entry = Iterables.getOnlyElement(scanner);
        assertEquals("10", entry.getValue().toString());

        ConditionalMutation cm8 = new ConditionalMutation("ACCUMULO-1000",
            new Condition("count", "comments").setIterators(aiConfig2, aiConfig3).setValue("35"));
        cm8.put("count", "comments", "9");
        assertEquals(Status.ACCEPTED, cw.write(cm8).getStatus());

        entry = Iterables.getOnlyElement(scanner);
        assertEquals("11", entry.getValue().toString());

        ConditionalMutation cm3 = new ConditionalMutation("ACCUMULO-1000",
            new Condition("count", "comments").setIterators(aiConfig2).setValue("33"));
        cm3.put("count", "comments", "3");

        ConditionalMutation cm4 = new ConditionalMutation("ACCUMULO-1001",
            new Condition("count", "comments").setIterators(aiConfig3).setValue("14"));
        cm4.put("count", "comments", "3");

        ConditionalMutation cm5 = new ConditionalMutation("ACCUMULO-1002",
            new Condition("count", "comments").setIterators(aiConfig3).setValue("10"));
        cm5.put("count", "comments", "3");

        Iterator<Result> results = cw.write(Arrays.asList(cm3, cm4, cm5).iterator());
        Map<String,Status> actual = new HashMap<>();

        while (results.hasNext()) {
          Result result = results.next();
          String k = new String(result.getMutation().getRow());
          assertFalse("Did not expect to see multiple resultus for the row: " + k,
              actual.containsKey(k));
          actual.put(k, result.getStatus());
        }

        Map<String,Status> expected = new HashMap<>();
        expected.put("ACCUMULO-1000", Status.ACCEPTED);
        expected.put("ACCUMULO-1001", Status.ACCEPTED);
        expected.put("ACCUMULO-1002", Status.REJECTED);
        assertEquals(expected, actual);
      }
    }
  }

  @Test
  public void testBatch() throws Exception {

    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      client.tableOperations().create(tableName);

      client.securityOperations().changeUserAuthorizations(getAdminPrincipal(),
          new Authorizations("A", "B"));

      ColumnVisibility cvab = new ColumnVisibility("A|B");

      ArrayList<ConditionalMutation> mutations = new ArrayList<>();

      ConditionalMutation cm0 =
          new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvab));
      cm0.put("name", "last", cvab, "doe");
      cm0.put("name", "first", cvab, "john");
      cm0.put("tx", "seq", cvab, "1");
      mutations.add(cm0);

      ConditionalMutation cm1 =
          new ConditionalMutation("59056", new Condition("tx", "seq").setVisibility(cvab));
      cm1.put("name", "last", cvab, "doe");
      cm1.put("name", "first", cvab, "jane");
      cm1.put("tx", "seq", cvab, "1");
      mutations.add(cm1);

      ConditionalMutation cm2 =
          new ConditionalMutation("19059", new Condition("tx", "seq").setVisibility(cvab));
      cm2.put("name", "last", cvab, "doe");
      cm2.put("name", "first", cvab, "jack");
      cm2.put("tx", "seq", cvab, "1");
      mutations.add(cm2);

      try (
          ConditionalWriter cw = client.createConditionalWriter(tableName,
              new ConditionalWriterConfig().setAuthorizations(new Authorizations("A")));
          Scanner scanner = client.createScanner(tableName, new Authorizations("A"))) {
        Iterator<Result> results = cw.write(mutations.iterator());
        int count = 0;
        while (results.hasNext()) {
          Result result = results.next();
          assertEquals(Status.ACCEPTED, result.getStatus());
          count++;
        }

        assertEquals(3, count);

        scanner.fetchColumn("tx", "seq");

        for (String row : new String[] {"99006", "59056", "19059"}) {
          scanner.setRange(new Range(row));
          Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
          assertEquals("1", entry.getValue().toString());
        }

        TreeSet<Text> splits = new TreeSet<>();
        splits.add(new Text("7"));
        splits.add(new Text("3"));
        client.tableOperations().addSplits(tableName, splits);

        mutations.clear();

        ConditionalMutation cm3 = new ConditionalMutation("99006",
            new Condition("tx", "seq").setVisibility(cvab).setValue("1"));
        cm3.put("name", "last", cvab, "Doe");
        cm3.put("tx", "seq", cvab, "2");
        mutations.add(cm3);

        ConditionalMutation cm4 =
            new ConditionalMutation("59056", new Condition("tx", "seq").setVisibility(cvab));
        cm4.put("name", "last", cvab, "Doe");
        cm4.put("tx", "seq", cvab, "1");
        mutations.add(cm4);

        ConditionalMutation cm5 = new ConditionalMutation("19059",
            new Condition("tx", "seq").setVisibility(cvab).setValue("2"));
        cm5.put("name", "last", cvab, "Doe");
        cm5.put("tx", "seq", cvab, "3");
        mutations.add(cm5);

        results = cw.write(mutations.iterator());
        int accepted = 0;
        int rejected = 0;
        while (results.hasNext()) {
          Result result = results.next();
          if (new String(result.getMutation().getRow()).equals("99006")) {
            assertEquals(Status.ACCEPTED, result.getStatus());
            accepted++;
          } else {
            assertEquals(Status.REJECTED, result.getStatus());
            rejected++;
          }
        }

        assertEquals("Expected only one accepted conditional mutation", 1, accepted);
        assertEquals("Expected two rejected conditional mutations", 2, rejected);

        for (String row : new String[] {"59056", "19059"}) {
          scanner.setRange(new Range(row));
          Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
          assertEquals("1", entry.getValue().toString());
        }

        scanner.setRange(new Range("99006"));
        Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
        assertEquals("2", entry.getValue().toString());

        scanner.clearColumns();
        scanner.fetchColumn("name", "last");
        entry = Iterables.getOnlyElement(scanner);
        assertEquals("Doe", entry.getValue().toString());
      }
    }
  }

  @Test
  public void testBigBatch() throws Exception {

    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      NewTableConfiguration ntc = new NewTableConfiguration().withSplits(nss("2", "4", "6"));
      client.tableOperations().create(tableName, ntc);

      sleepUninterruptibly(2, TimeUnit.SECONDS);

      int num = 100;

      ArrayList<byte[]> rows = new ArrayList<>(num);
      ArrayList<ConditionalMutation> cml = new ArrayList<>(num);

      Random r = new SecureRandom();
      byte[] e = new byte[0];

      for (int i = 0; i < num; i++) {
        rows.add(FastFormat.toZeroPaddedString(abs(r.nextLong()), 16, 16, e));
      }

      for (int i = 0; i < num; i++) {
        ConditionalMutation cm = new ConditionalMutation(rows.get(i), new Condition("meta", "seq"));

        cm.put("meta", "seq", "1");
        cm.put("meta", "tx", UUID.randomUUID().toString());

        cml.add(cm);
      }

      try (ConditionalWriter cw =
          client.createConditionalWriter(tableName, new ConditionalWriterConfig())) {

        Iterator<Result> results = cw.write(cml.iterator());

        int count = 0;

        // TODO check got each row back
        while (results.hasNext()) {
          Result result = results.next();
          assertEquals(Status.ACCEPTED, result.getStatus());
          count++;
        }

        assertEquals("Did not receive the expected number of results", num, count);

        ArrayList<ConditionalMutation> cml2 = new ArrayList<>(num);

        for (int i = 0; i < num; i++) {
          ConditionalMutation cm =
              new ConditionalMutation(rows.get(i), new Condition("meta", "seq").setValue("1"));

          cm.put("meta", "seq", "2");
          cm.put("meta", "tx", UUID.randomUUID().toString());

          cml2.add(cm);
        }

        count = 0;

        results = cw.write(cml2.iterator());

        while (results.hasNext()) {
          Result result = results.next();
          assertEquals(Status.ACCEPTED, result.getStatus());
          count++;
        }

        assertEquals("Did not receive the expected number of results", num, count);
      }
    }
  }

  @Test
  public void testBatchErrors() throws Exception {

    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      client.tableOperations().create(tableName);
      client.tableOperations().addConstraint(tableName, AlphaNumKeyConstraint.class.getName());
      client.tableOperations().clone(tableName, tableName + "_clone", true, new HashMap<>(),
          new HashSet<>());

      client.securityOperations().changeUserAuthorizations(getAdminPrincipal(),
          new Authorizations("A", "B"));

      ColumnVisibility cvaob = new ColumnVisibility("A|B");
      ColumnVisibility cvaab = new ColumnVisibility("A&B");

      switch ((new SecureRandom()).nextInt(3)) {
        case 1:
          client.tableOperations().addSplits(tableName, nss("6"));
          break;
        case 2:
          client.tableOperations().addSplits(tableName, nss("2", "95"));
          break;
      }

      ArrayList<ConditionalMutation> mutations = new ArrayList<>();

      ConditionalMutation cm0 =
          new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvaob));
      cm0.put("name+", "last", cvaob, "doe");
      cm0.put("name", "first", cvaob, "john");
      cm0.put("tx", "seq", cvaob, "1");
      mutations.add(cm0);

      ConditionalMutation cm1 =
          new ConditionalMutation("59056", new Condition("tx", "seq").setVisibility(cvaab));
      cm1.put("name", "last", cvaab, "doe");
      cm1.put("name", "first", cvaab, "jane");
      cm1.put("tx", "seq", cvaab, "1");
      mutations.add(cm1);

      ConditionalMutation cm2 =
          new ConditionalMutation("19059", new Condition("tx", "seq").setVisibility(cvaob));
      cm2.put("name", "last", cvaob, "doe");
      cm2.put("name", "first", cvaob, "jack");
      cm2.put("tx", "seq", cvaob, "1");
      mutations.add(cm2);

      ConditionalMutation cm3 = new ConditionalMutation("90909",
          new Condition("tx", "seq").setVisibility(cvaob).setValue("1"));
      cm3.put("name", "last", cvaob, "doe");
      cm3.put("name", "first", cvaob, "john");
      cm3.put("tx", "seq", cvaob, "2");
      mutations.add(cm3);

      try (
          ConditionalWriter cw = client.createConditionalWriter(tableName,
              new ConditionalWriterConfig().setAuthorizations(new Authorizations("A")));
          Scanner scanner = client.createScanner(tableName, new Authorizations("A"))) {
        Iterator<Result> results = cw.write(mutations.iterator());
        HashSet<String> rows = new HashSet<>();
        while (results.hasNext()) {
          Result result = results.next();
          String row = new String(result.getMutation().getRow());
          if (row.equals("19059")) {
            assertEquals(Status.ACCEPTED, result.getStatus());
          } else if (row.equals("59056")) {
            assertEquals(Status.INVISIBLE_VISIBILITY, result.getStatus());
          } else if (row.equals("99006")) {
            assertEquals(Status.VIOLATED, result.getStatus());
          } else if (row.equals("90909")) {
            assertEquals(Status.REJECTED, result.getStatus());
          }
          rows.add(row);
        }

        assertEquals(4, rows.size());

        scanner.fetchColumn("tx", "seq");
        Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
        assertEquals("1", entry.getValue().toString());
      }
    }
  }

  @Test
  public void testSameRow() throws Exception {
    // test multiple mutations for same row in same batch

    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String tableName = getUniqueNames(1)[0];

      client.tableOperations().create(tableName);

      try (ConditionalWriter cw =
          client.createConditionalWriter(tableName, new ConditionalWriterConfig())) {

        ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
        cm1.put("tx", "seq", "1");
        cm1.put("data", "x", "a");

        assertEquals(Status.ACCEPTED, cw.write(cm1).getStatus());

        ConditionalMutation cm2 =
            new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1"));
        cm2.put("tx", "seq", "2");
        cm2.put("data", "x", "b");

        ConditionalMutation cm3 =
            new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1"));
        cm3.put("tx", "seq", "2");
        cm3.put("data", "x", "c");

        ConditionalMutation cm4 =
            new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1"));
        cm4.put("tx", "seq", "2");
        cm4.put("data", "x", "d");

        Iterator<Result> results = cw.write(Arrays.asList(cm2, cm3, cm4).iterator());

        int accepted = 0;
        int rejected = 0;
        int total = 0;

        while (results.hasNext()) {
          Status status = results.next().getStatus();
          if (status == Status.ACCEPTED)
            accepted++;
          if (status == Status.REJECTED)
            rejected++;
          total++;
        }

        assertEquals("Expected one accepted result", 1, accepted);
        assertEquals("Expected two rejected results", 2, rejected);
        assertEquals("Expected three total results", 3, total);
      }
    }
  }

  private static class Stats {

    ByteSequence row = null;
    int seq;
    long sum;
    int[] data = new int[10];

    public Stats(Iterator<Entry<Key,Value>> iterator) {
      while (iterator.hasNext()) {
        Entry<Key,Value> entry = iterator.next();

        if (row == null)
          row = entry.getKey().getRowData();

        String cf = entry.getKey().getColumnFamilyData().toString();
        String cq = entry.getKey().getColumnQualifierData().toString();

        if (cf.equals("data")) {
          data[Integer.parseInt(cq)] = Integer.parseInt(entry.getValue().toString());
        } else if (cf.equals("meta")) {
          if (cq.equals("sum")) {
            sum = Long.parseLong(entry.getValue().toString());
          } else if (cq.equals("seq")) {
            seq = Integer.parseInt(entry.getValue().toString());
          }
        }
      }

      long sum2 = 0;

      for (int datum : data) {
        sum2 += datum;
      }

      assertEquals(sum2, sum);
    }

    public Stats(ByteSequence row) {
      this.row = row;
      for (int i = 0; i < data.length; i++) {
        this.data[i] = 0;
      }
      this.seq = -1;
      this.sum = 0;
    }

    void set(int index, int value) {
      sum -= data[index];
      sum += value;
      data[index] = value;
    }

    ConditionalMutation toMutation() {
      Condition cond = new Condition("meta", "seq");
      if (seq >= 0)
        cond.setValue(seq + "");

      ConditionalMutation cm = new ConditionalMutation(row, cond);

      cm.put("meta", "seq", (seq + 1) + "");
      cm.put("meta", "sum", (sum) + "");

      for (int i = 0; i < data.length; i++) {
        cm.put("data", i + "", data[i] + "");
      }

      return cm;
    }

    @Override
    public String toString() {
      return row + " " + seq + " " + sum;
    }
  }

  private static class MutatorTask implements Runnable {
    String tableName;
    ArrayList<ByteSequence> rows;
    ConditionalWriter cw;
    AccumuloClient client;
    AtomicBoolean failed;

    public MutatorTask(String tableName, AccumuloClient client, ArrayList<ByteSequence> rows,
        ConditionalWriter cw, AtomicBoolean failed) {
      this.tableName = tableName;
      this.rows = rows;
      this.client = client;
      this.cw = cw;
      this.failed = failed;
    }

    @Override
    public void run() {
      try (Scanner scanner =
          new IsolatedScanner(client.createScanner(tableName, Authorizations.EMPTY))) {
        Random rand = new SecureRandom();

        for (int i = 0; i < 20; i++) {
          int numRows = rand.nextInt(10) + 1;

          ArrayList<ByteSequence> changes = new ArrayList<>(numRows);
          ArrayList<ConditionalMutation> mutations = new ArrayList<>();

          for (int j = 0; j < numRows; j++)
            changes.add(rows.get(rand.nextInt(rows.size())));

          for (ByteSequence row : changes) {
            scanner.setRange(new Range(row.toString()));
            Stats stats = new Stats(scanner.iterator());
            stats.set(rand.nextInt(10), rand.nextInt(Integer.MAX_VALUE));
            mutations.add(stats.toMutation());
          }

          ArrayList<ByteSequence> changed = new ArrayList<>(numRows);
          Iterator<Result> results = cw.write(mutations.iterator());
          while (results.hasNext()) {
            Result result = results.next();
            changed.add(new ArrayByteSequence(result.getMutation().getRow()));
          }

          Collections.sort(changes);
          Collections.sort(changed);

          assertEquals(changes, changed);
        }
      } catch (Exception e) {
        log.error("{}", e.getMessage(), e);
        failed.set(true);
      }
    }
  }

  @Test
  public void testThreads() throws Exception {
    // test multiple threads using a single conditional writer

    String tableName = getUniqueNames(1)[0];
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {

      NewTableConfiguration ntc = new NewTableConfiguration();

      Random rand = new SecureRandom();

      switch (rand.nextInt(3)) {
        case 1:
          ntc = ntc.withSplits(nss("4"));
          break;
        case 2:
          ntc = ntc.withSplits(nss("3", "5"));
          break;
      }

      client.tableOperations().create(tableName, ntc);

      try (ConditionalWriter cw =
          client.createConditionalWriter(tableName, new ConditionalWriterConfig())) {

        ArrayList<ByteSequence> rows = new ArrayList<>();

        for (int i = 0; i < 1000; i++) {
          rows.add(new ArrayByteSequence(
              FastFormat.toZeroPaddedString(abs(rand.nextLong()), 16, 16, new byte[0])));
        }

        ArrayList<ConditionalMutation> mutations = new ArrayList<>();

        for (ByteSequence row : rows)
          mutations.add(new Stats(row).toMutation());

        ArrayList<ByteSequence> rows2 = new ArrayList<>();
        Iterator<Result> results = cw.write(mutations.iterator());
        while (results.hasNext()) {
          Result result = results.next();
          assertEquals(Status.ACCEPTED, result.getStatus());
          rows2.add(new ArrayByteSequence(result.getMutation().getRow()));
        }

        Collections.sort(rows);
        Collections.sort(rows2);

        assertEquals(rows, rows2);

        AtomicBoolean failed = new AtomicBoolean(false);

        ExecutorService tp = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 5; i++) {
          tp.submit(new MutatorTask(tableName, client, rows, cw, failed));
        }

        tp.shutdown();

        while (!tp.isTerminated()) {
          tp.awaitTermination(1, TimeUnit.MINUTES);
        }

        assertFalse("A MutatorTask failed with an exception", failed.get());
      }

      try (Scanner scanner = client.createScanner(tableName, Authorizations.EMPTY)) {
        RowIterator rowIter = new RowIterator(scanner);

        while (rowIter.hasNext()) {
          Iterator<Entry<Key,Value>> row = rowIter.next();
          new Stats(row);
        }
      }
    }
  }

  private SortedSet<Text> nss(String... splits) {
    TreeSet<Text> ret = new TreeSet<>();
    for (String split : splits)
      ret.add(new Text(split));

    return ret;
  }

  @Test
  public void testSecurity() throws Exception {
    // test against table user does not have read and/or write permissions for
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      String user = null;

      // Create a new user
      ClusterUser user1 = getUser(0);
      user = user1.getPrincipal();
      if (saslEnabled()) {
        client.securityOperations().createLocalUser(user, null);
      } else {
        client.securityOperations().createLocalUser(user, new PasswordToken(user1.getPassword()));
      }

      String[] tables = getUniqueNames(3);
      String table1 = tables[0], table2 = tables[1], table3 = tables[2];

      // Create three tables
      client.tableOperations().create(table1);
      client.tableOperations().create(table2);
      client.tableOperations().create(table3);

      // Grant R on table1, W on table2, R/W on table3
      client.securityOperations().grantTablePermission(user, table1, TablePermission.READ);
      client.securityOperations().grantTablePermission(user, table2, TablePermission.WRITE);
      client.securityOperations().grantTablePermission(user, table3, TablePermission.READ);
      client.securityOperations().grantTablePermission(user, table3, TablePermission.WRITE);

      ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
      cm1.put("tx", "seq", "1");
      cm1.put("data", "x", "a");

      try (
          AccumuloClient client2 =
              Accumulo.newClient().from(client.properties()).as(user, user1.getToken()).build();
          ConditionalWriter cw1 =
              client2.createConditionalWriter(table1, new ConditionalWriterConfig());
          ConditionalWriter cw2 =
              client2.createConditionalWriter(table2, new ConditionalWriterConfig());
          ConditionalWriter cw3 =
              client2.createConditionalWriter(table3, new ConditionalWriterConfig())) {

        // Should be able to conditional-update a table we have R/W on
        assertEquals(Status.ACCEPTED, cw3.write(cm1).getStatus());

        // Conditional-update to a table we only have read on should fail
        try {
          Status status = cw1.write(cm1).getStatus();
          fail("Expected exception writing conditional mutation to table"
              + " the user doesn't have write access to, Got status: " + status);
        } catch (AccumuloSecurityException ase) {

        }

        // Conditional-update to a table we only have writer on should fail
        try {
          Status status = cw2.write(cm1).getStatus();
          fail("Expected exception writing conditional mutation to table"
              + " the user doesn't have read access to. Got status: " + status);
        } catch (AccumuloSecurityException ase) {

        }
      }
    }
  }

  @Test
  public void testTimeout() throws Exception {
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {

      String table = getUniqueNames(1)[0];

      client.tableOperations().create(table);

      try (
          ConditionalWriter cw = client.createConditionalWriter(table,
              new ConditionalWriterConfig().setTimeout(3, TimeUnit.SECONDS));
          Scanner scanner = client.createScanner(table, Authorizations.EMPTY)) {

        ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
        cm1.put("tx", "seq", "1");
        cm1.put("data", "x", "a");

        assertEquals(cw.write(cm1).getStatus(), Status.ACCEPTED);

        IteratorSetting is = new IteratorSetting(5, SlowIterator.class);
        SlowIterator.setSeekSleepTime(is, 5000);

        ConditionalMutation cm2 = new ConditionalMutation("r1",
            new Condition("tx", "seq").setValue("1").setIterators(is));
        cm2.put("tx", "seq", "2");
        cm2.put("data", "x", "b");

        assertEquals(cw.write(cm2).getStatus(), Status.UNKNOWN);

        for (Entry<Key,Value> entry : scanner) {
          String cf = entry.getKey().getColumnFamilyData().toString();
          String cq = entry.getKey().getColumnQualifierData().toString();
          String val = entry.getValue().toString();

          if (cf.equals("tx") && cq.equals("seq"))
            assertEquals("Unexpected value in tx:seq", "1", val);
          else if (cf.equals("data") && cq.equals("x"))
            assertEquals("Unexpected value in data:x", "a", val);
          else
            fail("Saw unexpected column family and qualifier: " + entry);
        }

        ConditionalMutation cm3 =
            new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1"));
        cm3.put("tx", "seq", "2");
        cm3.put("data", "x", "b");

        assertEquals(cw.write(cm3).getStatus(), Status.ACCEPTED);
      }
    }
  }

  @Test
  public void testDeleteTable() throws Exception {
    String table = getUniqueNames(1)[0];
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {

      try {
        client.createConditionalWriter(table, new ConditionalWriterConfig());
        fail("Creating conditional writer for table that doesn't exist should fail");
      } catch (TableNotFoundException e) {}

      client.tableOperations().create(table);

      try (ConditionalWriter cw =
          client.createConditionalWriter(table, new ConditionalWriterConfig())) {

        client.tableOperations().delete(table);

        ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
        cm1.put("tx", "seq", "1");
        cm1.put("data", "x", "a");

        Result result = cw.write(cm1);

        try {
          Status status = result.getStatus();
          fail("Expected exception writing conditional mutation to deleted table. Got status: "
              + status);
        } catch (AccumuloException ae) {
          assertEquals(TableDeletedException.class, ae.getCause().getClass());
        }
      }
    }
  }

  @Test
  public void testOffline() throws Exception {
    String table = getUniqueNames(1)[0];
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {

      client.tableOperations().create(table);

      try (ConditionalWriter cw =
          client.createConditionalWriter(table, new ConditionalWriterConfig())) {

        client.tableOperations().offline(table, true);

        ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
        cm1.put("tx", "seq", "1");
        cm1.put("data", "x", "a");

        Result result = cw.write(cm1);

        try {
          Status status = result.getStatus();
          fail("Expected exception writing conditional mutation to offline table. Got status: "
              + status);
        } catch (AccumuloException ae) {
          assertEquals(TableOfflineException.class, ae.getCause().getClass());
        }

        try {
          client.createConditionalWriter(table, new ConditionalWriterConfig());
          fail("Expected exception creating conditional writer to offline table");
        } catch (TableOfflineException e) {}
      }
    }
  }

  @Test
  public void testError() throws Exception {
    String table = getUniqueNames(1)[0];
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {

      client.tableOperations().create(table);

      try (ConditionalWriter cw =
          client.createConditionalWriter(table, new ConditionalWriterConfig())) {

        IteratorSetting iterSetting = new IteratorSetting(5, BadIterator.class);

        ConditionalMutation cm1 =
            new ConditionalMutation("r1", new Condition("tx", "seq").setIterators(iterSetting));
        cm1.put("tx", "seq", "1");
        cm1.put("data", "x", "a");

        Result result = cw.write(cm1);

        try {
          Status status = result.getStatus();
          fail("Expected exception using iterator which throws an error, Got status: " + status);
        } catch (AccumuloException ae) {

        }

      }
    }
  }

  @Test(expected = IllegalArgumentException.class)
  public void testNoConditions() throws AccumuloException, AccumuloSecurityException,
      TableExistsException, TableNotFoundException {
    String table = getUniqueNames(1)[0];
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {

      client.tableOperations().create(table);

      try (ConditionalWriter cw =
          client.createConditionalWriter(table, new ConditionalWriterConfig())) {

        ConditionalMutation cm1 = new ConditionalMutation("r1");
        cm1.put("tx", "seq", "1");
        cm1.put("data", "x", "a");

        cw.write(cm1);
      }
    }
  }

  @Test
  public void testTrace() throws Exception {
    // Need to add a getClientConfig() to AccumuloCluster
    Process tracer = null;
    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
      MiniAccumuloClusterImpl mac = getCluster();
      if (!client.tableOperations().exists("trace")) {
        tracer = mac.exec(TraceServer.class).getProcess();
        while (!client.tableOperations().exists("trace")) {
          sleepUninterruptibly(1, TimeUnit.SECONDS);
        }
      }

      String tableName = getUniqueNames(1)[0];
      client.tableOperations().create(tableName);

      TraceUtil.enableClientTraces("localhost", "testTrace", mac.getClientProperties());
      sleepUninterruptibly(1, TimeUnit.SECONDS);
      long rootTraceId;
      try (TraceScope root = Trace.startSpan("traceTest", Sampler.ALWAYS); ConditionalWriter cw =
          client.createConditionalWriter(tableName, new ConditionalWriterConfig())) {
        rootTraceId = root.getSpan().getTraceId();

        // mutation conditional on column tx:seq not exiting
        ConditionalMutation cm0 = new ConditionalMutation("99006", new Condition("tx", "seq"));
        cm0.put("name", "last", "doe");
        cm0.put("name", "first", "john");
        cm0.put("tx", "seq", "1");
        assertEquals(Status.ACCEPTED, cw.write(cm0).getStatus());
      }

      try (Scanner scanner = client.createScanner("trace", Authorizations.EMPTY)) {
        scanner.setRange(new Range(new Text(Long.toHexString(rootTraceId))));
        loop: while (true) {
          final StringBuilder finalBuffer = new StringBuilder();
          int traceCount = TraceDump.printTrace(scanner, line -> {
            try {
              finalBuffer.append(line).append("\n");
            } catch (Exception ex) {
              throw new RuntimeException(ex);
            }
          });
          String traceOutput = finalBuffer.toString();
          log.info("Trace output:" + traceOutput);
          if (traceCount > 0) {
            String[] parts = ("traceTest, startScan,startConditionalUpdate,conditionalUpdate"
                + ",Check conditions,apply conditional mutations").split(",");
            int lastPos = 0;
            for (String part : parts) {
              log.info("Looking in trace output for '" + part + "'");
              int pos = traceOutput.indexOf(part);
              if (pos == -1) {
                log.info("Trace output doesn't contain '" + part + "'");
                Thread.sleep(1000);
                break loop;
              }
              assertTrue("Did not find '" + part + "' in output", pos > 0);
              assertTrue("'" + part + "' occurred earlier than the previous element unexpectedly",
                  pos > lastPos);
              lastPos = pos;
            }
            break;
          } else {
            log.info("Ignoring trace output as traceCount not greater than zero: " + traceCount);
            Thread.sleep(1000);
          }
        }
        if (tracer != null) {
          tracer.destroy();
        }
      }
    }
  }
}