/**
 * Adds the mutations to labels region and set the results to the finalOpStatus. finalOpStatus
 * might have some entries in it where the OpStatus is FAILURE. We will leave those and set in
 * others in the order.
 * @param mutations
 * @param finalOpStatus
 * @return whether we need a ZK update or not.
 */
private boolean mutateLabelsRegion(List<Mutation> mutations, OperationStatus[] finalOpStatus)
    throws IOException {
  OperationStatus[] opStatus = this.labelsRegion.batchMutate(mutations
    .toArray(new Mutation[mutations.size()]));
  int i = 0;
  boolean updateZk = false;
  for (OperationStatus status : opStatus) {
    // Update the zk when atleast one of the mutation was added successfully.
    updateZk = updateZk || (status.getOperationStatusCode() == OperationStatusCode.SUCCESS);
    for (; i < finalOpStatus.length; i++) {
      if (finalOpStatus[i] == null) {
        finalOpStatus[i] = status;
        break;
      }
    }
  }
  return updateZk;
}
 

@Override
public OperationStatus[] setAuths(byte[] user, List<byte[]> authLabels) throws IOException {
  assert labelsRegion != null;
  OperationStatus[] finalOpStatus = new OperationStatus[authLabels.size()];
  Put p = new Put(user);
  CellBuilder builder = CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
  for (byte[] auth : authLabels) {
    p.add(builder.clear()
        .setRow(p.getRow())
        .setFamily(LABELS_TABLE_FAMILY)
        .setQualifier(auth)
        .setTimestamp(p.getTimestamp())
        .setType(Cell.Type.Put)
        .setValue(DUMMY_VALUE)
        .build());
  }
  this.labelsRegion.put(p);
  // This is a testing impl and so not doing any caching
  for (int i = 0; i < authLabels.size(); i++) {
    finalOpStatus[i] = new OperationStatus(OperationStatusCode.SUCCESS);
  }
  return finalOpStatus;
}
 

@Override
public boolean processPutStatus(MutationStatus operationStatus) throws IOException{
    OperationStatus opStat = ((HMutationStatus)operationStatus).unwrapDelegate();
     switch (opStat.getOperationStatusCode()) {
        case NOT_RUN:
            throw new IOException("Could not acquire Lock");
        case BAD_FAMILY:
            throw new NoSuchColumnFamilyException(opStat.getExceptionMsg());
        case SANITY_CHECK_FAILURE:
            throw new IOException("Sanity Check failure:" + opStat.getExceptionMsg());
        case FAILURE:
            throw new IOException(opStat.getExceptionMsg());
        default:
            return true;
     }
}
 

@Override
public OperationStatus[] addLabels(List<byte[]> labels) throws IOException {
  assert labelsRegion != null;
  OperationStatus[] finalOpStatus = new OperationStatus[labels.size()];
  List<Mutation> puts = new ArrayList<>(labels.size());
  int i = 0;
  ExtendedCellBuilder builder = ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
  for (byte[] label : labels) {
    String labelStr = Bytes.toString(label);
    if (this.labelsCache.getLabelOrdinal(labelStr) > 0) {
      finalOpStatus[i] = new OperationStatus(OperationStatusCode.FAILURE,
          new LabelAlreadyExistsException("Label '" + labelStr + "' already exists"));
    } else {
      byte[] row = Bytes.toBytes(ordinalCounter.get());
      Put p = new Put(row);
      p.add(builder.clear()
            .setRow(row)
            .setFamily(LABELS_TABLE_FAMILY)
            .setQualifier(LABEL_QUALIFIER)
            .setTimestamp(p.getTimestamp())
            .setType(Type.Put)
            .setValue(label)
            .setTags(TagUtil.fromList(Arrays.asList(LABELS_TABLE_TAGS)))
            .build());
      if (LOG.isDebugEnabled()) {
        LOG.debug("Adding the label " + labelStr);
      }
      puts.add(p);
      ordinalCounter.incrementAndGet();
    }
    i++;
  }
  if (mutateLabelsRegion(puts, finalOpStatus)) {
    updateZk(true);
  }
  return finalOpStatus;
}
 

@Override
public OperationStatus[] setAuths(byte[] user, List<byte[]> authLabels) throws IOException {
  assert labelsRegion != null;
  OperationStatus[] finalOpStatus = new OperationStatus[authLabels.size()];
  List<Mutation> puts = new ArrayList<>(authLabels.size());
  int i = 0;
  ExtendedCellBuilder builder = ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
  for (byte[] auth : authLabels) {
    String authStr = Bytes.toString(auth);
    int labelOrdinal = this.labelsCache.getLabelOrdinal(authStr);
    if (labelOrdinal == 0) {
      // This label is not yet added. 1st this should be added to the system
      finalOpStatus[i] = new OperationStatus(OperationStatusCode.FAILURE,
          new InvalidLabelException("Label '" + authStr + "' doesn't exists"));
    } else {
      byte[] row = Bytes.toBytes(labelOrdinal);
      Put p = new Put(row);
      p.add(builder.clear()
          .setRow(row)
          .setFamily(LABELS_TABLE_FAMILY)
          .setQualifier(user)
          .setTimestamp(p.getTimestamp())
          .setType(Cell.Type.Put)
          .setValue(DUMMY_VALUE)
          .setTags(TagUtil.fromList(Arrays.asList(LABELS_TABLE_TAGS)))
          .build());
      puts.add(p);
    }
    i++;
  }
  if (mutateLabelsRegion(puts, finalOpStatus)) {
    updateZk(false);
  }
  return finalOpStatus;
}
 

@Override
public OperationStatus[] clearAuths(byte[] user, List<byte[]> authLabels) throws IOException {
  assert labelsRegion != null;
  OperationStatus[] finalOpStatus = new OperationStatus[authLabels.size()];
  List<String> currentAuths;
  if (AuthUtil.isGroupPrincipal(Bytes.toString(user))) {
    String group = AuthUtil.getGroupName(Bytes.toString(user));
    currentAuths = this.getGroupAuths(new String[]{group}, true);
  }
  else {
    currentAuths = this.getUserAuths(user, true);
  }
  List<Mutation> deletes = new ArrayList<>(authLabels.size());
  int i = 0;
  for (byte[] authLabel : authLabels) {
    String authLabelStr = Bytes.toString(authLabel);
    if (currentAuths.contains(authLabelStr)) {
      int labelOrdinal = this.labelsCache.getLabelOrdinal(authLabelStr);
      assert labelOrdinal > 0;
      Delete d = new Delete(Bytes.toBytes(labelOrdinal));
      d.addColumns(LABELS_TABLE_FAMILY, user);
      deletes.add(d);
    } else {
      // This label is not set for the user.
      finalOpStatus[i] = new OperationStatus(OperationStatusCode.FAILURE,
          new InvalidLabelException("Label '" + authLabelStr + "' is not set for the user "
              + Bytes.toString(user)));
    }
    i++;
  }
  if (mutateLabelsRegion(deletes, finalOpStatus)) {
    updateZk(false);
  }
  return finalOpStatus;
}
 

@Override
public void preBatchMutate(final ObserverContext<RegionCoprocessorEnvironment> c,
                           final MiniBatchOperationInProgress<Mutation> miniBatchOp)
    throws IOException {
  if (ops.incrementAndGet() % 20000 == 0) {
    LOG.info("Wrote " + ops.get() + " times in region " + regionName);
  }

  for (int i = 0; i < miniBatchOp.size(); i++) {
    miniBatchOp.setOperationStatus(i,
        new OperationStatus(HConstants.OperationStatusCode.SUCCESS));
  }
  c.bypass();
}
 

@Override
public OperationStatus[] addLabels(List<byte[]> labels) throws IOException {
  // Not doing specific label add. We will just add labels in Mutation
  // visibility expression as it
  // is along with every cell.
  OperationStatus[] status = new OperationStatus[labels.size()];
  for (int i = 0; i < labels.size(); i++) {
    status[i] = new OperationStatus(OperationStatusCode.SUCCESS);
  }
  return status;
}
 

@Override
public OperationStatus[] clearAuths(byte[] user, List<byte[]> authLabels) throws IOException {
  assert labelsRegion != null;
  OperationStatus[] finalOpStatus = new OperationStatus[authLabels.size()];
  List<String> currentAuths;
  if (AuthUtil.isGroupPrincipal(Bytes.toString(user))) {
    String group = AuthUtil.getGroupName(Bytes.toString(user));
    currentAuths = this.getGroupAuths(new String[]{group}, true);
  }
  else {
    currentAuths = this.getUserAuths(user, true);
  }
  Delete d = new Delete(user);
  int i = 0;
  for (byte[] authLabel : authLabels) {
    String authLabelStr = Bytes.toString(authLabel);
    if (currentAuths.contains(authLabelStr)) {
      d.addColumns(LABELS_TABLE_FAMILY, authLabel);
    } else {
      // This label is not set for the user.
      finalOpStatus[i] = new OperationStatus(OperationStatusCode.FAILURE,
          new InvalidLabelException("Label '" + authLabelStr + "' is not set for the user "
              + Bytes.toString(user)));
    }
    i++;
  }
  this.labelsRegion.delete(d);
  // This is a testing impl and so not doing any caching
  for (i = 0; i < authLabels.size(); i++) {
    if (finalOpStatus[i] == null) {
      finalOpStatus[i] = new OperationStatus(OperationStatusCode.SUCCESS);
    }
  }
  return finalOpStatus;
}
 

@Override
public MutationStatus failure(Throwable t){
    if (t instanceof IOException) {
        return new HMutationStatus(new ExtendedOperationStatus(HConstants.OperationStatusCode.FAILURE, (IOException) t));
    } else {
        return new HMutationStatus(new OperationStatus(HConstants.OperationStatusCode.FAILURE, t.getMessage()));
    }
}
 

/**
 * {@inheritDoc}
 */
@Override
public OperationStatus[] put(Pair<Put, Integer>[] putsAndLocks)
    throws IOException {
  for (Pair<Put, Integer> pair : putsAndLocks) {
    updateIndexes(pair.getFirst(), pair.getSecond());
  }
  return super.put(putsAndLocks);
}
 

@Override
public MutationStatus failure(String message){
    return new HMutationStatus(new OperationStatus(HConstants.OperationStatusCode.FAILURE,message));
}
 

public HMutationStatus(OperationStatus delegate){
    this.delegate=delegate;
}
 

public void set(OperationStatus delegate){
    this.delegate = delegate;
}
 

public OperationStatus unwrapDelegate(){
    return delegate;
}
 

/**
 * Adds the set of labels into the system.
 * @param labels
 *          Labels to add to the system.
 * @return OperationStatus for each of the label addition
 */
OperationStatus[] addLabels(List<byte[]> labels) throws IOException;
 

/**
 * Sets given labels globally authorized for the user.
 * @param user
 *          The authorizing user
 * @param authLabels
 *          Labels which are getting authorized for the user
 * @return OperationStatus for each of the label auth addition
 */
OperationStatus[] setAuths(byte[] user, List<byte[]> authLabels) throws IOException;
 

/**
 * Removes given labels from user's globally authorized list of labels.
 * @param user
 *          The user whose authorization to be removed
 * @param authLabels
 *          Labels which are getting removed from authorization set
 * @return OperationStatus for each of the label auth removal
 */
OperationStatus[] clearAuths(byte[] user, List<byte[]> authLabels) throws IOException;