package com.github.ambry.coordinator;

import com.github.ambry.clustermap.ReplicaId;
import com.github.ambry.shared.ConnectionPool;
import com.github.ambry.shared.BlobId;
import com.github.ambry.shared.DeleteRequest;
import com.github.ambry.shared.RequestOrResponse;
import com.github.ambry.shared.ServerErrorCode;
import com.github.ambry.shared.Response;
import com.github.ambry.shared.DeleteResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.DataInputStream;
import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;


/**
 * Performs a delete operation by sending and receiving delete requests until operation is complete or has failed.
 */
final public class DeleteOperation extends Operation {
  private Logger logger = LoggerFactory.getLogger(getClass());

  public DeleteOperation(String datacenterName, ConnectionPool connectionPool, ExecutorService requesterPool,
      OperationContext oc, BlobId blobId, long operationTimeoutMs)
      throws CoordinatorException {
    super(datacenterName, connectionPool, requesterPool, oc, blobId, operationTimeoutMs,
        new AllInParallelOperationPolicy(datacenterName, blobId.getPartition()));
  }

  @Override
  protected OperationRequest makeOperationRequest(ReplicaId replicaId) {
    DeleteRequest deleteRequest = new DeleteRequest(context.getCorrelationId(), context.getClientId(), blobId);
    return new DeleteOperationRequest(connectionPool, responseQueue, context, blobId, replicaId, deleteRequest);
  }

  @Override
  protected boolean processResponseError(ReplicaId replicaId, ServerErrorCode serverErrorCode)
      throws CoordinatorException {
    switch (serverErrorCode) {
      case No_Error:
      case Blob_Deleted:
        return true;
      // Cannot delete if blob is not found
      case Blob_Not_Found:
        return false;
      case Blob_Expired:
      default:
        CoordinatorException e = new CoordinatorException("Server returned unexpected error for DeleteOperation.",
            CoordinatorError.UnexpectedInternalError);
        logger.error("{} DeleteResponse for BlobId {} received from ReplicaId {} had unexpected error code {}: {}",
            context, blobId, replicaId, serverErrorCode, e);
        throw e;
    }
  }
}

final class DeleteOperationRequest extends OperationRequest {
  protected DeleteOperationRequest(ConnectionPool connectionPool, BlockingQueue<OperationResponse> responseQueue,
      OperationContext context, BlobId blobId, ReplicaId replicaId, RequestOrResponse request) {
    super(connectionPool, responseQueue, context, blobId, replicaId, request);
  }

  @Override
  protected void markRequest()
      throws CoordinatorException {
    context.getCoordinatorMetrics().getRequestMetrics(replicaId.getDataNodeId()).deleteBlobRequestRate.mark();
  }

  @Override
  protected void updateRequest(long durationInMs)
      throws CoordinatorException {
    context.getCoordinatorMetrics().
        getRequestMetrics(replicaId.getDataNodeId()).deleteBlobRequestLatencyInMs.update(durationInMs);
  }

  @Override
  protected Response getResponse(DataInputStream dataInputStream)
      throws IOException {
    return DeleteResponse.readFrom(dataInputStream);
  }
}



@@ -844,16 +844,217 @@
 lass());
+%0A  private int blobNotFoundCount;%0A  // Number of replicas in the partition. This is used to set threshold to determine blob not found (all replicas%0A  // must reply).%0A  private final int replicaIdCount;
 %0A%0A  publ
@@ -1430,16 +1430,121 @@
 ion()));
+%0A%0A    this.replicaIdCount = blobId.getPartition().getReplicaIds().size();%0A    this.blobNotFoundCount = 0;
 %0A  %7D%0A%0A  
@@ -2141,16 +2141,16 @@
 t found%0A
-
       ca
@@ -2168,16 +2168,338 @@
 _Found:%0A
+        blobNotFoundCount++;%0A        if (blobNotFoundCount == replicaIdCount) %7B%0A          String message = %22Blob not found : blobNotFoundCount == replicaIdCount == %22 + blobNotFoundCount + %22.%22;%0A          logger.trace(message);%0A          throw new CoordinatorException(message, CoordinatorError.BlobDoesNotExist);%0A        %7D%0A
         

Change how coordinator deleteBlob handles 'not found' from all servers