Java 类org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException 实例源码

private int doPuts(int maxOps, final HTable... tables) throws Exception {
  int count = 0;
  try {
    while (count < maxOps) {
      Put put = new Put(Bytes.toBytes("row-" + count));
      put.addColumn(FAMILY, QUALIFIER, Bytes.toBytes("data-" + count));
      for (final HTable table : tables) {
        table.put(put);
      }
      count += tables.length;
    }
  } catch (RetriesExhaustedWithDetailsException e) {
    for (Throwable t : e.getCauses()) {
      if (!(t instanceof ThrottlingException)) {
        throw e;
      }
    }
    LOG.error("put failed after nRetries=" + count, e);
  }
  return count;
}

private void recordFailure(final Mutation m, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to mutate: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
      + exceptionInfo);
}

private void recordFailure(final Table table, final Put put, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow()) + "; errors: "
      + exceptionInfo);
}

@Override
public Put processEvent(Event event)
        throws EventProcessingException, RetriesExhaustedWithDetailsException, InterruptedIOException {
    Set<Entry<String, String>> kvp = event.getHeaders().entrySet();
    // Iterator<Entry<String, String>> itr = kvp.iterator();
    rowKey = getRowKey(constructDefaultRowKey(event.getHeaders(), 0), event.getHeaders());
    Put putReq = new Put(rowKey);
    // while(itr.hasNext()){
    // Entry<String, String> entry = itr.next();
    // if(entry.getValue()!=null){
    //
    // }
    // }
    putReq.add(MarauderParserConstants.MARAUDER_CF_HEADERS, "v".getBytes(),
            event.getHeaders().toString().getBytes());
    // if(event.getBody()!=null){
    // putReq.add(MarauderParserConstants.MARAUDER_CF_MESSAGE,
    // MarauderParserConstants.MARAUDER_KEY_MESSAGE,
    // compressEventBody(event.getBody()));
    // }
    performExtendedProcessing(putReq, event);
    return putReq;
}

private void doMutation(final Mutation mutation) throws RetriesExhaustedWithDetailsException {
  Long sequenceId = null;
  try {
    // registerOperationWithHeapSize() waits until both the memory and rpc count maximum
    // requirements are achieved.
    sequenceId = sizeManager.registerOperationWithHeapSize(mutation.heapSize());
  } catch (InterruptedException e) {
    Thread.currentThread().interrupt();
  }
  AccountingFutureCallback callback = new AccountingFutureCallback(mutation, sequenceId);

  // TODO: Consider adding the callback in another executor for the blocking call. There are some
  // concerns running this callback on the rpc threadpool. This callback requires locks on
  // resources. If they cannot be acquired, the callback would block future completions.
  Futures.addCallback(batchExecutor.issueRequest(mutation), callback);
}

public void insert(Table table, Put put, long keyBase) {
  long start = System.currentTimeMillis();
  try {
    put = (Put) dataGenerator.beforeMutate(keyBase, put);
    table.put(put);
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException)e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      StringWriter stackWriter = new StringWriter();
      PrintWriter pw = new PrintWriter(stackWriter);
      e.printStackTrace(pw);
      pw.flush();
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
        + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow())
        + "; errors: " + exceptionInfo);
  }
}

private void recordFailure(final Mutation m, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to mutate: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
      + exceptionInfo);
}

private void recordFailure(final Table table, final Put put, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow()) + "; errors: "
      + exceptionInfo);
}

public void insert(HTable table, Put put, long keyBase) {
  long start = System.currentTimeMillis();
  try {
    put = (Put) dataGenerator.beforeMutate(keyBase, put);
    table.put(put);
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException)e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      StringWriter stackWriter = new StringWriter();
      PrintWriter pw = new PrintWriter(stackWriter);
      e.printStackTrace(pw);
      pw.flush();
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
        + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow())
        + "; errors: " + exceptionInfo);
  }
}

private void recordFailure(final Mutation m, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to mutate: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
      + exceptionInfo);
}

private void recordFailure(final HTable table, final Put put, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow()) + "; errors: "
      + exceptionInfo);
}

public static void handleHBaseException(
  RetriesExhaustedWithDetailsException rex,
  Record record,
  Map<String, Record> rowKeyToRecord,
  ErrorRecordHandler errorRecordHandler
) throws StageException {
  for (int i = 0; i < rex.getNumExceptions(); i++) {
    if (rex.getCause(i) instanceof NoSuchColumnFamilyException) {
      Row r = rex.getRow(i);
      Record errorRecord = record != null ? record : rowKeyToRecord.get(Bytes.toString(r.getRow()));
      OnRecordErrorException exception =
        new OnRecordErrorException(errorRecord, Errors.HBASE_10,
          getErrorDescription(rex.getCause(i), r, rex.getHostnamePort(i)));
      errorRecordHandler.onError(exception);
    } else {
      // If at least 1 non NoSuchColumnFamilyException exception,
      // consider as stage exception
      throw new StageException(Errors.HBASE_02, rex);
    }
  }
}

/**
 * Do the changes and handle the pool
 * @param tableName table to insert into
 * @param allRows list of actions
 * @throws IOException
 */
protected void batch(TableName tableName, Collection<List<Row>> allRows) throws IOException {
  if (allRows.isEmpty()) {
    return;
  }
  Table table = null;
  try {
    Connection connection = getConnection();
    table = connection.getTable(tableName);
    for (List<Row> rows : allRows) {
      table.batch(rows, null);
    }
  } catch (RetriesExhaustedWithDetailsException rewde) {
    for (Throwable ex : rewde.getCauses()) {
      if (ex instanceof TableNotFoundException) {
        throw new TableNotFoundException("'"+tableName+"'");
      }
    }
  } catch (InterruptedException ix) {
    throw (InterruptedIOException) new InterruptedIOException().initCause(ix);
  } finally {
    if (table != null) {
      table.close();
    }
  }
}

public void insert(Table table, Put put, long keyBase) {
  long start = System.currentTimeMillis();
  try {
    put = (Put) dataGenerator.beforeMutate(keyBase, put);
    table.put(put);
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException)e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      StringWriter stackWriter = new StringWriter();
      PrintWriter pw = new PrintWriter(stackWriter);
      e.printStackTrace(pw);
      pw.flush();
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
        + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow())
        + "; errors: " + exceptionInfo);
  }
}

private void recordFailure(final Mutation m, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to mutate: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
      + exceptionInfo);
}

private void recordFailure(final Table table, final Put put, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow()) + "; errors: "
      + exceptionInfo);
}

public void insert(HTable table, Put put, long keyBase) {
  long start = System.currentTimeMillis();
  try {
    put = (Put) dataGenerator.beforeMutate(keyBase, put);
    table.put(put);
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException)e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      StringWriter stackWriter = new StringWriter();
      PrintWriter pw = new PrintWriter(stackWriter);
      e.printStackTrace(pw);
      pw.flush();
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
        + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow())
        + "; errors: " + exceptionInfo);
  }
}

private void recordFailure(final Mutation m, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to mutate: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
      + exceptionInfo);
}

private void recordFailure(final HTable table, final Put put, final long keyBase,
    final long start, IOException e) {
  failedKeySet.add(keyBase);
  String exceptionInfo;
  if (e instanceof RetriesExhaustedWithDetailsException) {
    RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
    exceptionInfo = aggEx.getExhaustiveDescription();
  } else {
    StringWriter stackWriter = new StringWriter();
    PrintWriter pw = new PrintWriter(stackWriter);
    e.printStackTrace(pw);
    pw.flush();
    exceptionInfo = StringUtils.stringifyException(e);
  }
  LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start)
      + "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow()) + "; errors: "
      + exceptionInfo);
}

public void insert(HTable table, Put put, long keyBase) {
  long start = System.currentTimeMillis();
  try {
    table.put(put);
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException)e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      StringWriter stackWriter = new StringWriter();
      PrintWriter pw = new PrintWriter(stackWriter);
      e.printStackTrace(pw);
      pw.flush();
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to insert: " + keyBase + " after " + (System.currentTimeMillis() - start) +
      "ms; region information: " + getRegionDebugInfoSafe(table, put.getRow()) + "; errors: "
        + exceptionInfo);
  }
}

/**
 * Test that constraints will fail properly
 * @throws Exception
 */
@SuppressWarnings("unchecked")
@Test(timeout = 60000)
public void testConstraintFails() throws Exception {

  // create the table
  // it would be nice if this was also a method on the util
  HTableDescriptor desc = new HTableDescriptor(tableName);
  for (byte[] family : new byte[][] { dummy, test }) {
    desc.addFamily(new HColumnDescriptor(family));
  }

  // add a constraint that is sure to fail
  Constraints.add(desc, AllFailConstraint.class);

  util.getHBaseAdmin().createTable(desc);
  Table table = new HTable(util.getConfiguration(), tableName);

  // test that we do fail on violation
  Put put = new Put(row1);
  put.add(dummy, new byte[0], "fail".getBytes());
  LOG.warn("Doing put in table");
  try {
    table.put(put);
    fail("This put should not have suceeded - AllFailConstraint was not run!");
  } catch (RetriesExhaustedWithDetailsException e) {
    List<Throwable> causes = e.getCauses();
    assertEquals(
        "More than one failure cause - should only be the failure constraint exception",
        1, causes.size());
    Throwable t = causes.get(0);
    assertEquals(ConstraintException.class, t.getClass());
  }
  table.close();
}

public void mutate(Table table, Mutation m,
    long keyBase, byte[] row, byte[] cf, byte[] q, byte[] v) {
  long start = System.currentTimeMillis();
  try {
    m = dataGenerator.beforeMutate(keyBase, m);
    if (m instanceof Increment) {
      table.increment((Increment)m);
    } else if (m instanceof Append) {
      table.append((Append)m);
    } else if (m instanceof Put) {
      table.checkAndPut(row, cf, q, v, (Put)m);
    } else if (m instanceof Delete) {
      table.checkAndDelete(row, cf, q, v, (Delete)m);
    } else {
      throw new IllegalArgumentException(
        "unsupported mutation " + m.getClass().getSimpleName());
    }
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    if (ignoreNonceConflicts && (e instanceof OperationConflictException)) {
      LOG.info("Detected nonce conflict, ignoring: " + e.getMessage());
      totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
      return;
    }
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to mutate: " + keyBase + " after " +
        (System.currentTimeMillis() - start) +
      "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
        + exceptionInfo);
  }
}

public void mutate(Table table, Mutation m,
    long keyBase, byte[] row, byte[] cf, byte[] q, byte[] v) {
  long start = System.currentTimeMillis();
  try {
    m = dataGenerator.beforeMutate(keyBase, m);
    if (m instanceof Increment) {
      table.increment((Increment)m);
    } else if (m instanceof Append) {
      table.append((Append)m);
    } else if (m instanceof Put) {
      table.checkAndPut(row, cf, q, v, (Put)m);
    } else if (m instanceof Delete) {
      table.checkAndDelete(row, cf, q, v, (Delete)m);
    } else {
      throw new IllegalArgumentException(
        "unsupported mutation " + m.getClass().getSimpleName());
    }
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException)e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      StringWriter stackWriter = new StringWriter();
      PrintWriter pw = new PrintWriter(stackWriter);
      e.printStackTrace(pw);
      pw.flush();
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to mutate: " + keyBase + " after " + (System.currentTimeMillis() - start) +
      "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
        + exceptionInfo);
  }
}

private HTable getTable() throws IOException {
  HTable table = tables.get();
  if (table == null) {
    table = new HTable(conf, tableName) {
      @Override
      public synchronized void flushCommits() throws RetriesExhaustedWithDetailsException, InterruptedIOException {
        super.flushCommits();
      }
    };
    table.setAutoFlushTo(autoFlush);
    pool.add(table); //keep track
    tables.set(table);
  }
  return table;
}

@Override
    public Put processEvent(Event event) throws EventProcessingException, RetriesExhaustedWithDetailsException, InterruptedIOException {
        time = Integer.parseInt(event.getHeaders().get(MarauderParserConstants.MARAUDER_KEY_TIMESTAMP), 16);
        rowKey = getRowKey(constructDefaultRowKey(event.getHeaders(), time), 
                event.getHeaders());        
//      System.out.println(new Date((long)time*1000));
        Put putReq = new Put(rowKey);
        offset = TimeUtils.getWindowOffsetTime(time, windowSize);
        column = new byte[6];
        System.arraycopy(ByteUtils.shortToByteMSB(offset), 0, column, 0, 2);
        System.arraycopy(ByteUtils.intToByteMSB(random.nextInt(1000000)), 0, column, 2, 4);
        putReq.add(MarauderParserConstants.MARAUDER_CF_HEADERS, column, event.getHeaders().toString().getBytes());
        performExtendedProcessing(putReq, event);
        return putReq;
    }

/**
 * Test that constraints will fail properly
 * @throws Exception
 */
@SuppressWarnings("unchecked")
@Test(timeout = 60000)
public void testConstraintFails() throws Exception {

  // create the table
  // it would be nice if this was also a method on the util
  HTableDescriptor desc = new HTableDescriptor(tableName);
  for (byte[] family : new byte[][] { dummy, test }) {
    desc.addFamily(new HColumnDescriptor(family));
  }

  // add a constraint that is sure to fail
  Constraints.add(desc, AllFailConstraint.class);

  util.getHBaseAdmin().createTable(desc);
  HTable table = new HTable(util.getConfiguration(), tableName);
  table.setAutoFlush(true);

  // test that we do fail on violation
  Put put = new Put(row1);
  put.add(dummy, new byte[0], "fail".getBytes());
  LOG.warn("Doing put in table");
  try {
    table.put(put);
    fail("This put should not have suceeded - AllFailConstraint was not run!");
  } catch (RetriesExhaustedWithDetailsException e) {
    List<Throwable> causes = e.getCauses();
    assertEquals(
        "More than one failure cause - should only be the failure constraint exception",
        1, causes.size());
    Throwable t = causes.get(0);
    assertEquals(ConstraintException.class, t.getClass());
  }
  table.close();
}

public void flushTuples()
{
  try {
    store.flushTables();
  } catch (InterruptedIOException | RetriesExhaustedWithDetailsException e) {
    throw Throwables.propagate(e);
  }
}

public void flushTables() throws InterruptedIOException, RetriesExhaustedWithDetailsException
{
  if (table != null) {
    flushTable(table);
  }
  for (Map.Entry<String, HTable> entry : tableCache.asMap().entrySet()) {
    flushTable(entry.getValue());
  }
}

@Override
public void commitTransaction()
{
  try {
    flushTables();
  } catch (InterruptedIOException | RetriesExhaustedWithDetailsException e) {
    throw Throwables.propagate(e);
  }
}

/**
 * Create a {@link RetriesExhaustedWithDetailsException} if there were any async exceptions and
 * send it to the {@link org.apache.hadoop.hbase.client.BufferedMutator.ExceptionListener}.
 */
@VisibleForTesting
void handleExceptions() throws RetriesExhaustedWithDetailsException {
  if (hasExceptions.get()) {
    ArrayList<MutationException> mutationExceptions = null;
    synchronized (globalExceptions) {
      mutationExceptions = new ArrayList<>(globalExceptions);
      globalExceptions.clear();
      hasExceptions.set(false);
    }

    if (mutationExceptions.isEmpty()) {
      // This should never happen.
      return;
    }

    List<Throwable> problems = new ArrayList<>(mutationExceptions.size());
    ArrayList<String> hostnames = new ArrayList<>(mutationExceptions.size());
    List<Row> failedMutations = new ArrayList<>(mutationExceptions.size());

    for (MutationException mutationException : mutationExceptions) {
      problems.add(mutationException.throwable);
      failedMutations.add(mutationException.mutation);
      hostnames.add(host);
    }

    RetriesExhaustedWithDetailsException exception = new RetriesExhaustedWithDetailsException(
        problems, failedMutations, hostnames);
    exceptionListener.onException(exception, this);
  }
}

private void setup() {
  setup(new BufferedMutator.ExceptionListener() {
    @Override
    public void onException(RetriesExhaustedWithDetailsException exception,
        BufferedMutator mutator) throws RetriesExhaustedWithDetailsException {
      throw exception;
    }
  });
}

@Test
public void testException() {
  underTest.hasExceptions.set(true);
  underTest.globalExceptions.add(
      new BigtableBufferedMutator.MutationException(null, new Exception()));

  try {
    underTest.handleExceptions();
    Assert.fail("expected RetriesExhaustedWithDetailsException");
  } catch (RetriesExhaustedWithDetailsException expected) {
    // Expected
  }
}

@Test(expected = RetriesExhaustedWithDetailsException.class)
@Category(KnownGap.class)
public void testIOExceptionOnFailedPut() throws Exception {
  Table table = getConnection().getTable(TABLE_NAME);
  byte[] rowKey = Bytes.toBytes("testrow-" + RandomStringUtils.randomAlphanumeric(8));
  byte[] badfamily = Bytes.toBytes("badcolumnfamily-" + RandomStringUtils.randomAlphanumeric(8));
  byte[] qualifier = Bytes.toBytes("testQualifier-" + RandomStringUtils.randomAlphanumeric(8));
  byte[] value = Bytes.toBytes("testValue-" + RandomStringUtils.randomAlphanumeric(8));
  Put put = new Put(rowKey);
  put.addColumn(badfamily, qualifier, value);
  table.put(put);
}

@Test
@Category(KnownGap.class)
public void testAtomicPut() throws Exception {
  Table table = getConnection().getTable(TABLE_NAME);
  byte[] rowKey = Bytes.toBytes("testrow-" + RandomStringUtils.randomAlphanumeric(8));
  byte[] goodQual = Bytes.toBytes("testQualifier-" + RandomStringUtils.randomAlphanumeric(8));
  byte[] goodValue = Bytes.toBytes("testValue-" + RandomStringUtils.randomAlphanumeric(8));
  byte[] badQual = Bytes.toBytes("testQualifier-" + RandomStringUtils.randomAlphanumeric(8));
  byte[] badValue = Bytes.toBytes("testValue-" + RandomStringUtils.randomAlphanumeric(8));
  byte[] badfamily = Bytes.toBytes("badcolumnfamily-" + RandomStringUtils.randomAlphanumeric(8));
  Put put = new Put(rowKey);
  put.addColumn(COLUMN_FAMILY, goodQual, goodValue);
  put.addColumn(badfamily, badQual, badValue);
  RetriesExhaustedWithDetailsException thrownException = null;
  try {
    table.put(put);
  } catch (RetriesExhaustedWithDetailsException e) {
    thrownException = e;
  }
  Assert.assertNotNull("Exception should have been thrown", thrownException);
  Assert.assertEquals("Expecting one exception", 1, thrownException.getNumExceptions());
  Assert.assertArrayEquals("Row key", rowKey, thrownException.getRow(0).getRow());
  Assert.assertTrue("Cause: NoSuchColumnFamilyException",
      thrownException.getCause(0) instanceof NoSuchColumnFamilyException);

  Get get = new Get(rowKey);
  Result result = table.get(get);
  Assert.assertEquals("Atomic behavior means there should be nothing here", 0, result.size());
  table.close();
}

/**
 * Test that constraints will fail properly
 * @throws Exception
 */
@SuppressWarnings("unchecked")
@Test(timeout = 60000)
public void testConstraintFails() throws Exception {

  // create the table
  // it would be nice if this was also a method on the util
  HTableDescriptor desc = new HTableDescriptor(tableName);
  for (byte[] family : new byte[][] { dummy, test }) {
    desc.addFamily(new HColumnDescriptor(family));
  }

  // add a constraint that is sure to fail
  Constraints.add(desc, AllFailConstraint.class);

  util.getHBaseAdmin().createTable(desc);
  Table table = new HTable(util.getConfiguration(), tableName);

  // test that we do fail on violation
  Put put = new Put(row1);
  put.add(dummy, new byte[0], "fail".getBytes());
  LOG.warn("Doing put in table");
  try {
    table.put(put);
    fail("This put should not have suceeded - AllFailConstraint was not run!");
  } catch (RetriesExhaustedWithDetailsException e) {
    List<Throwable> causes = e.getCauses();
    assertEquals(
        "More than one failure cause - should only be the failure constraint exception",
        1, causes.size());
    Throwable t = causes.get(0);
    assertEquals(ConstraintException.class, t.getClass());
  }
  table.close();
}

private Table doPutsWithoutVisibility(TableName tableName) throws IOException,
    InterruptedIOException, RetriesExhaustedWithDetailsException, InterruptedException {
  Admin hBaseAdmin = TEST_UTIL.getHBaseAdmin();
  HColumnDescriptor colDesc = new HColumnDescriptor(fam);
  colDesc.setMaxVersions(5);
  HTableDescriptor desc = new HTableDescriptor(tableName);
  desc.addFamily(colDesc);
  hBaseAdmin.createTable(desc);
  List<Put> puts = new ArrayList<>();
  Put put = new Put(Bytes.toBytes("row1"));
  put.add(fam, qual, 123l, value);
  puts.add(put);

  put = new Put(Bytes.toBytes("row1"));
  put.add(fam, qual, 124l, value);
  puts.add(put);

  put = new Put(Bytes.toBytes("row1"));
  put.add(fam, qual, 125l, value);
  puts.add(put);

  put = new Put(Bytes.toBytes("row1"));
  put.add(fam, qual, 126l, value);
  puts.add(put);

  put = new Put(Bytes.toBytes("row1"));
  put.add(fam, qual, 127l, value);
  puts.add(put);

  Table table = new HTable(conf, tableName);
  table.put(puts);

  TEST_UTIL.getHBaseAdmin().flush(tableName);

  put = new Put(Bytes.toBytes("row2"));
  put.add(fam, qual, 127l, value);
  table.put(put);

  return table;
}

public void mutate(Table table, Mutation m,
    long keyBase, byte[] row, byte[] cf, byte[] q, byte[] v) {
  long start = System.currentTimeMillis();
  try {
    m = dataGenerator.beforeMutate(keyBase, m);
    if (m instanceof Increment) {
      table.increment((Increment)m);
    } else if (m instanceof Append) {
      table.append((Append)m);
    } else if (m instanceof Put) {
      table.checkAndPut(row, cf, q, v, (Put)m);
    } else if (m instanceof Delete) {
      table.checkAndDelete(row, cf, q, v, (Delete)m);
    } else {
      throw new IllegalArgumentException(
        "unsupported mutation " + m.getClass().getSimpleName());
    }
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    if (ignoreNonceConflicts && (e instanceof OperationConflictException)) {
      LOG.info("Detected nonce conflict, ignoring: " + e.getMessage());
      totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
      return;
    }
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to mutate: " + keyBase + " after " +
        (System.currentTimeMillis() - start) +
      "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
        + exceptionInfo);
  }
}

public void mutate(Table table, Mutation m,
    long keyBase, byte[] row, byte[] cf, byte[] q, byte[] v) {
  long start = System.currentTimeMillis();
  try {
    m = dataGenerator.beforeMutate(keyBase, m);
    if (m instanceof Increment) {
      table.increment((Increment)m);
    } else if (m instanceof Append) {
      table.append((Append)m);
    } else if (m instanceof Put) {
      table.checkAndPut(row, cf, q, v, (Put)m);
    } else if (m instanceof Delete) {
      table.checkAndDelete(row, cf, q, v, (Delete)m);
    } else {
      throw new IllegalArgumentException(
        "unsupported mutation " + m.getClass().getSimpleName());
    }
    totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
  } catch (IOException e) {
    failedKeySet.add(keyBase);
    String exceptionInfo;
    if (e instanceof RetriesExhaustedWithDetailsException) {
      RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException)e;
      exceptionInfo = aggEx.getExhaustiveDescription();
    } else {
      StringWriter stackWriter = new StringWriter();
      PrintWriter pw = new PrintWriter(stackWriter);
      e.printStackTrace(pw);
      pw.flush();
      exceptionInfo = StringUtils.stringifyException(e);
    }
    LOG.error("Failed to mutate: " + keyBase + " after " + (System.currentTimeMillis() - start) +
      "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
        + exceptionInfo);
  }
}

/**
 * Test that constraints will fail properly
 * @throws Exception
 */
@SuppressWarnings("unchecked")
@Test(timeout = 60000)
public void testConstraintFails() throws Exception {

  // create the table
  // it would be nice if this was also a method on the util
  HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
  for (byte[] family : new byte[][] { dummy, test }) {
    desc.addFamily(new HColumnDescriptor(family));
  }

  // add a constraint that is sure to fail
  Constraints.add(desc, AllFailConstraint.class);

  util.getHBaseAdmin().createTable(desc);
  HTable table = new HTable(util.getConfiguration(), tableName);

  // test that we do fail on violation
  Put put = new Put(row1);
  put.add(dummy, new byte[0], "fail".getBytes());
  LOG.warn("Doing put in table");
  try {
    table.put(put);
    fail("This put should not have suceeded - AllFailConstraint was not run!");
  } catch (RetriesExhaustedWithDetailsException e) {
    List<Throwable> causes = e.getCauses();
    assertEquals(
        "More than one failure cause - should only be the failure constraint exception",
        1, causes.size());
    Throwable t = causes.get(0);
    assertEquals(ConstraintException.class, t.getClass());
  }
  table.close();
}

public void mutate(HTable table, Mutation m,
     long keyBase, byte[] row, byte[] cf, byte[] q, byte[] v) {
   long start = System.currentTimeMillis();
   try {
     m = dataGenerator.beforeMutate(keyBase, m);
     if (m instanceof Increment) {
       table.increment((Increment)m);
     } else if (m instanceof Append) {
       table.append((Append)m);
     } else if (m instanceof Put) {
       table.checkAndPut(row, cf, q, v, (Put)m);
     } else if (m instanceof Delete) {
       table.checkAndDelete(row, cf, q, v, (Delete)m);
     } else {
       throw new IllegalArgumentException(
         "unsupported mutation " + m.getClass().getSimpleName());
     }
     totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
   } catch (IOException e) {
     if (ignoreNonceConflicts && (e instanceof OperationConflictException)) {
       LOG.info("Detected nonce conflict, ignoring: " + e.getMessage());
totalOpTimeMs.addAndGet(System.currentTimeMillis() - start);
       return;
     }
     failedKeySet.add(keyBase);
     String exceptionInfo;
     if (e instanceof RetriesExhaustedWithDetailsException) {
       RetriesExhaustedWithDetailsException aggEx = (RetriesExhaustedWithDetailsException) e;
       exceptionInfo = aggEx.getExhaustiveDescription();
     } else {
       exceptionInfo = StringUtils.stringifyException(e);
     }
     LOG.error("Failed to mutate: " + keyBase + " after " +
         (System.currentTimeMillis() - start) +
       "ms; region information: " + getRegionDebugInfoSafe(table, m.getRow()) + "; errors: "
         + exceptionInfo);
   }
 }