/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.proxy.s3;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.client.file.FileSystem;
import alluxio.client.file.URIStatus;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AccessControlException;
import alluxio.exception.AlluxioException;
import alluxio.exception.DirectoryNotEmptyException;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.grpc.Bits;
import alluxio.grpc.CreateDirectoryPOptions;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.ListStatusPOptions;
import alluxio.grpc.PMode;
import alluxio.grpc.SetAttributePOptions;
import alluxio.proto.journal.File;
import alluxio.s3.DeleteObjectsRequest;
import alluxio.s3.DeleteObjectsResult;
import alluxio.s3.ListAllMyBucketsResult;
import alluxio.s3.ListBucketOptions;
import alluxio.s3.ListBucketResult;
import alluxio.s3.S3AuditContext;
import alluxio.s3.S3Constants;
import alluxio.s3.S3ErrorCode;
import alluxio.s3.S3Exception;
import alluxio.s3.TaggingData;

import com.fasterxml.jackson.dataformat.xml.XmlMapper;
import com.google.common.base.Preconditions;
import com.google.common.net.InetAddresses;
import com.google.protobuf.ByteString;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.stream.Collectors;
import javax.ws.rs.core.Response;

/**
 * S3 Tasks to handle bucket level or global level request.
 * (only bucket name or no bucket name is provided)
 */
public class S3BucketTask extends S3BaseTask {
  private static final Logger LOG = LoggerFactory.getLogger(S3BucketTask.class);

  protected S3BucketTask(S3Handler handler, OpType opType) {
    super(handler, opType);
  }

  @Override
  public Response continueTask() {
    return S3RestUtils.call(mHandler.getBucket(), () -> {
      throw new S3Exception(S3ErrorCode.NOT_IMPLEMENTED);
    });
  }

  /**
   * Factory for getting a S3BucketTask type task.
   */
  public static final class Factory {
    /**
     * Marshall the request and create corresponding bucket level S3 task.
     * @param handler
     * @return S3BucketTask
     */
    public static S3BucketTask create(S3Handler handler) {
      switch (handler.getHTTPVerb()) {
        case "GET":
          if (StringUtils.isEmpty(handler.getBucket())) {
            return new ListBucketsTask(handler, OpType.ListBuckets);
          } else if (handler.getQueryParameter("tagging") != null) {
            return new GetBucketTaggingTask(handler, OpType.GetBucketTagging);
          } else if (handler.getQueryParameter("uploads") != null) {
            return new ListMultipartUploadsTask(handler, OpType.ListMultipartUploads);
          } else {
            return new ListObjectsTask(handler, OpType.ListObjects);
          }
        case "PUT":
          if (handler.getQueryParameter("tagging") != null) {
            return new PutBucketTaggingTask(handler, OpType.PutBucketTagging);
          } else {
            return new CreateBucketTask(handler, OpType.CreateBucket);
          }
        case "POST":
          if (handler.getQueryParameter("delete") != null) {
            return new DeleteObjectsTask(handler, OpType.DeleteObjects);
          }
          break;
        case "HEAD":
          if (!StringUtils.isEmpty(handler.getBucket())) {
            return new HeadBucketTask(handler, OpType.HeadBucket);
          }
          break;
        case "DELETE":
          if (handler.getQueryParameter("tagging") != null) {
            return new DeleteBucketTaggingTask(handler, OpType.DeleteBucketTagging);
          } else {
            return new DeleteBucketTask(handler, OpType.DeleteBucket);
          }
        default:
          break;
      }
      return new S3BucketTask(handler, OpType.Unsupported);
    }
  }

  private static class ListBucketsTask extends S3BucketTask {
    protected ListBucketsTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    @Override
    public Response continueTask() {
      return S3RestUtils.call(S3Constants.EMPTY, () -> {
        final String user = mHandler.getUser();

        List<URIStatus> objects;
        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, null, null)) {
          try {
            objects = mHandler.getMetaFS().listStatus(new AlluxioURI("/"));
          } catch (AlluxioException | IOException e) {
            if (e instanceof AccessControlException) {
              auditContext.setAllowed(false);
            }
            auditContext.setSucceeded(false);
            throw S3RestUtils.toBucketS3Exception(e, "/");
          }

          final List<URIStatus> buckets = objects.stream()
                  .filter((uri) -> !uri.getName().equals(S3Constants.S3_METADATA_ROOT_DIR))
                  // debatable (?) potentially breaks backcompat(?)
                  .filter(URIStatus::isFolder)
                  .collect(Collectors.toList());
          buckets.forEach(
              (uri) -> mHandler.BUCKET_PATH_CACHE.put(uri.getPath(), true));
          return new ListAllMyBucketsResult(buckets);
        }
      });
    }
  } // end of ListBucketsTask

  private static class GetBucketTaggingTask extends S3BucketTask {
    protected GetBucketTaggingTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {

        String path = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + mHandler.getBucket());
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(
            mHandler.getUser(), mHandler.getMetaFS());

        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, mHandler.getBucket(), null)) {
          S3RestUtils.checkPathIsAlluxioDirectory(userFs, path, auditContext,
              mHandler.BUCKET_PATH_CACHE);
          AlluxioURI uri = new AlluxioURI(path);
          try {
            TaggingData tagData = S3RestUtils.deserializeTags(userFs.getStatus(uri).getXAttr());
            LOG.debug("GetBucketTagging tagData={}", tagData);
            return tagData != null ? tagData : new TaggingData();
          } catch (Exception e) {
            throw S3RestUtils.toBucketS3Exception(e, mHandler.getBucket(), auditContext);
          }
        }
      });
    }
  } // end of GetBucketTaggingTask

  private static class ListMultipartUploadsTask extends S3BucketTask {

    protected ListMultipartUploadsTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {
        final String bucket = mHandler.getBucket();
        Preconditions.checkNotNull(bucket, "required 'bucket' parameter is missing");

        String path = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + bucket);
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(user, mHandler.getMetaFS());

        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, mHandler.getBucket(), null)) {
          S3RestUtils.checkPathIsAlluxioDirectory(userFs, path, auditContext,
              mHandler.BUCKET_PATH_CACHE);
          try {
            List<URIStatus> children = mHandler.getMetaFS().listStatus(new AlluxioURI(
                    S3RestUtils.MULTIPART_UPLOADS_METADATA_DIR));
            // GetListing doesn't contain the xattr info, so get status for MPU files.
            List<URIStatus> mpuList = new ArrayList<>(children.size());
            for (URIStatus status : children) {
              URIStatus filemeta = mHandler.getMetaFS().getStatus(new AlluxioURI(status.getPath()));
              mpuList.add(filemeta);
            }
            return ListMultipartUploadsResult.buildFromStatuses(bucket, mpuList);
          } catch (Exception e) {
            throw S3RestUtils.toBucketS3Exception(e, bucket, auditContext);
          }
        }
      });
    }
  } // end of ListMultipartUploadsTask

  private static class ListObjectsTask extends S3BucketTask {
    protected ListObjectsTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    private String normalizeS3Prefix(String prefix, char delimiter) {
      if (prefix != null) {
        int pos = prefix.lastIndexOf(delimiter);
        if (pos >= 0) {
          return prefix.substring(0, pos + 1);
        }
      }
      return S3Constants.EMPTY;
    }

    private String parsePathWithDelimiter(String bucketPath, String prefix, String delimiter)
            throws S3Exception {
      // TODO(czhu): allow non-"/" delimiters
      // Alluxio only support use / as delimiter
      if (!delimiter.equals(AlluxioURI.SEPARATOR)) {
        throw new S3Exception(bucketPath, new S3ErrorCode(
                S3ErrorCode.PRECONDITION_FAILED.getCode(),
                "Alluxio S3 API only support / as delimiter.",
                S3ErrorCode.PRECONDITION_FAILED.getStatus()));
      }
      char delim = AlluxioURI.SEPARATOR.charAt(0);
      String normalizedBucket =
              bucketPath.replace(S3Constants.BUCKET_SEPARATOR, AlluxioURI.SEPARATOR);
      String normalizedPrefix = normalizeS3Prefix(prefix, delim);

      if (!normalizedPrefix.isEmpty() && !normalizedPrefix.startsWith(AlluxioURI.SEPARATOR)) {
        normalizedPrefix = AlluxioURI.SEPARATOR + normalizedPrefix;
      }
      return normalizedBucket + normalizedPrefix;
    }

    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {
        String path = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + mHandler.getBucket());
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(user, mHandler.getMetaFS());

        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, mHandler.getBucket(), null)) {
          S3RestUtils.checkPathIsAlluxioDirectory(userFs, path, auditContext,
              mHandler.BUCKET_PATH_CACHE);
          String markerParam = mHandler.getQueryParameter("marker");
          String maxKeysParam = mHandler.getQueryParameter("max-keys");
          String prefixParam = mHandler.getQueryParameter("prefix");
          String delimiterParam = mHandler.getQueryParameter("delimiter");
          String encodingTypeParam = mHandler.getQueryParameter("encoding-type");
          String listTypeParam = mHandler.getQueryParameter("list-type");
          String continuationTokenParam = mHandler.getQueryParameter("continuation-token");
          String startAfterParam = mHandler.getQueryParameter("start-after");

          int maxKeys = maxKeysParam == null
              ? ListBucketOptions.DEFAULT_MAX_KEYS : Integer.parseInt(maxKeysParam);
          Integer listType = listTypeParam == null ? null : Integer.parseInt(listTypeParam);
          ListBucketOptions listBucketOptions = ListBucketOptions.defaults()
                  .setMarker(markerParam)
                  .setPrefix(prefixParam)
                  .setMaxKeys(maxKeys)
                  .setDelimiter(delimiterParam)
                  .setEncodingType(encodingTypeParam)
                  .setListType(listType)
                  .setContinuationToken(continuationTokenParam)
                  .setStartAfter(startAfterParam);

          List<URIStatus> children;
          try {
            // TODO(czhu): allow non-"/" delimiters by parsing the prefix & delimiter pair to
            //             determine what directory to list the contents of
            //             only list the direct children if delimiter is not null
            if (StringUtils.isNotEmpty(delimiterParam)) {
              if (prefixParam == null) {
                path = parsePathWithDelimiter(path, S3Constants.EMPTY, delimiterParam);
              } else {
                path = parsePathWithDelimiter(path, prefixParam, delimiterParam);
              }
              children = userFs.listStatus(new AlluxioURI(path));
            } else {
              if (prefixParam != null) {
                path = parsePathWithDelimiter(path, prefixParam, AlluxioURI.SEPARATOR);
              }
              ListStatusPOptions options = ListStatusPOptions.newBuilder()
                  .setRecursive(true).build();
              children = userFs.listStatus(new AlluxioURI(path), options);
            }
          } catch (IOException | AlluxioException e) {
            auditContext.setSucceeded(false);
            throw S3RestUtils.toBucketS3Exception(e, mHandler.getBucket());
          }
          return new ListBucketResult(
                  mHandler.getBucket(),
                  children,
                  listBucketOptions);
        } // end try-with-resources block
      });
    }
  } // end of ListObjectsTask

  private static class PutBucketTaggingTask extends S3BucketTask {

    protected PutBucketTaggingTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    @Override
    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(user, mHandler.getMetaFS());
        String bucketPath = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + mHandler.getBucket());
        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), mHandler.getUser(), mHandler.getBucket(), null)) {
          S3RestUtils.checkPathIsAlluxioDirectory(mHandler.getMetaFS(), bucketPath, auditContext,
              mHandler.BUCKET_PATH_CACHE);
          try {
            TaggingData tagData = new XmlMapper().readerFor(TaggingData.class)
                    .readValue(mHandler.getInputStream());
            LOG.debug("PutBucketTagging tagData={}", tagData);
            Map<String, ByteString> xattrMap = new HashMap<>();
            xattrMap.put(S3Constants.TAGGING_XATTR_KEY, TaggingData.serialize(tagData));
            SetAttributePOptions attrPOptions = SetAttributePOptions.newBuilder()
                    .putAllXattr(xattrMap)
                    .setXattrUpdateStrategy(File.XAttrUpdateStrategy.UNION_REPLACE)
                    .build();
            userFs.setAttribute(new AlluxioURI(bucketPath), attrPOptions);
          } catch (IOException e) {
            if (e.getCause() instanceof S3Exception) {
              throw S3RestUtils.toBucketS3Exception((S3Exception) e.getCause(), bucketPath,
                      auditContext);
            }
            auditContext.setSucceeded(false);
            throw new S3Exception(e, bucketPath, S3ErrorCode.MALFORMED_XML);
          } catch (Exception e) {
            throw S3RestUtils.toBucketS3Exception(e, bucketPath, auditContext);
          }
          return Response.Status.OK;
        }
      });
    }
  } // end of PutBucketTaggingTask

  private static class CreateBucketTask extends S3BucketTask {
    protected CreateBucketTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    @Override
    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(user, mHandler.getMetaFS());
        String bucketPath = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + mHandler.getBucket());
        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, mHandler.getBucket(), null)) {
          if (S3Handler.BUCKET_NAMING_RESTRICTION_ENABLED) {
            Matcher m = S3Handler.BUCKET_ADJACENT_DOTS_DASHES_PATTERN.matcher(mHandler.getBucket());
            while (m.find()) {
              if (!m.group().equals("--")) {
                auditContext.setSucceeded(false);
                throw new S3Exception(mHandler.getBucket(), S3ErrorCode.INVALID_BUCKET_NAME);
              }
            }
            if (!S3Handler.BUCKET_VALID_NAME_PATTERN.matcher(mHandler.getBucket()).matches()
                || S3Handler.BUCKET_INVALIDATION_PREFIX_PATTERN.matcher(mHandler.getBucket())
                .matches()
                || S3Handler.BUCKET_INVALID_SUFFIX_PATTERN.matcher(mHandler.getBucket()).matches()
                || InetAddresses.isInetAddress(mHandler.getBucket())) {
              auditContext.setSucceeded(false);
              throw new S3Exception(mHandler.getBucket(), S3ErrorCode.INVALID_BUCKET_NAME);
            }
          }
          try {
            URIStatus status = mHandler.getMetaFS().getStatus(new AlluxioURI(bucketPath));
            if (status.isFolder()) {
              if (status.getOwner().equals(user)) {
                // Silently swallow CreateBucket calls on existing buckets for this user
                // - S3 clients may prepend PutObject requests with CreateBucket calls instead of
                //   calling HeadBucket to ensure that the bucket exists
                mHandler.BUCKET_PATH_CACHE.put(bucketPath, true);
                return Response.Status.OK;
              }
              // Otherwise, this bucket is owned by a different user
              throw new S3Exception(S3ErrorCode.BUCKET_ALREADY_EXISTS);
            }
            // Otherwise, that path exists in Alluxio but is not a directory
            auditContext.setSucceeded(false);
            throw new InvalidPathException("A file already exists at bucket path " + bucketPath);
          } catch (FileDoesNotExistException e) {
            // do nothing, we will create the directory below
          } catch (Exception e) {
            throw S3RestUtils.toBucketS3Exception(e, bucketPath, auditContext);
          }

          // These permission bits will be inherited by all objects/folders created within
          // the bucket; we don't support custom bucket/object ACLs at the moment
          CreateDirectoryPOptions options =
                  CreateDirectoryPOptions.newBuilder()
                          .setMode(PMode.newBuilder()
                                  .setOwnerBits(Bits.ALL)
                                  .setGroupBits(Bits.ALL)
                                  .setOtherBits(Bits.NONE))
                          .setWriteType(S3RestUtils.getS3WriteType())
                          .build();
          try {
            mHandler.getMetaFS().createDirectory(new AlluxioURI(bucketPath), options);
            SetAttributePOptions attrPOptions = SetAttributePOptions.newBuilder()
                    .setOwner(user)
                    .build();
            mHandler.getMetaFS().setAttribute(new AlluxioURI(bucketPath), attrPOptions);
          } catch (Exception e) {
            throw S3RestUtils.toBucketS3Exception(e, bucketPath, auditContext);
          }
          mHandler.BUCKET_PATH_CACHE.put(bucketPath, true);
          return Response.Status.OK;
        }
      });
    }
  } // end of CreateBucketTask

  private static class DeleteObjectsTask extends S3BucketTask {

    protected DeleteObjectsTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(user, mHandler.getMetaFS());
        String bucketPath = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + mHandler.getBucket());
        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, mHandler.getBucket(), null)) {
          try {
            DeleteObjectsRequest request = new XmlMapper().readerFor(DeleteObjectsRequest.class)
                    .readValue(mHandler.getInputStream());
            List<DeleteObjectsRequest.DeleteObject> objs = request.getToDelete();
            List<DeleteObjectsResult.DeletedObject> success = new ArrayList<>();
            List<DeleteObjectsResult.ErrorObject> errored = new ArrayList<>();
            objs.sort(Comparator.comparingInt(x -> -1 * x.getKey().length()));
            objs.forEach(obj -> {
              try {
                AlluxioURI uri = new AlluxioURI(bucketPath
                        + AlluxioURI.SEPARATOR + obj.getKey());
                DeletePOptions options = DeletePOptions.newBuilder().build();
                userFs.delete(uri, options);
                DeleteObjectsResult.DeletedObject del = new DeleteObjectsResult.DeletedObject();
                del.setKey(obj.getKey());
                success.add(del);
              } catch (FileDoesNotExistException | DirectoryNotEmptyException e) {
              /*
              FDNE - delete on FDNE should be counted as a success, as there's nothing to do
              DNE - s3 has no concept dirs - if it _is_ a dir, nothing to delete.
               */
                DeleteObjectsResult.DeletedObject del = new DeleteObjectsResult.DeletedObject();
                del.setKey(obj.getKey());
                success.add(del);
              } catch (IOException | AlluxioException e) {
                DeleteObjectsResult.ErrorObject err = new DeleteObjectsResult.ErrorObject();
                err.setKey(obj.getKey());
                err.setMessage(e.getMessage());
                errored.add(err);
              }
            });

            DeleteObjectsResult result = new DeleteObjectsResult();
            if (!request.getQuiet()) {
              result.setDeleted(success);
            }
            result.setErrored(errored);
            return result;
          } catch (IOException e) {
            LOG.debug("Failed to parse DeleteObjects request:", e);
            auditContext.setSucceeded(false);
            return Response.Status.BAD_REQUEST;
          }
        }
      });
    }
  } // end of DeleteObjectsTask

  private static class HeadBucketTask extends S3BucketTask {
    protected HeadBucketTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    @Override
    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {
        String bucketPath = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + mHandler.getBucket());
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(user, mHandler.getMetaFS());

        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, mHandler.getBucket(), null)) {
          S3RestUtils.checkPathIsAlluxioDirectory(userFs, bucketPath, auditContext,
              mHandler.BUCKET_PATH_CACHE);
        }
        return Response.ok().build();
      });
    }
  } // end of HeadBucketTask

  private static class DeleteBucketTaggingTask extends S3BucketTask {

    protected DeleteBucketTaggingTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    @Override
    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(user, mHandler.getMetaFS());
        String bucketPath = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + mHandler.getBucket());
        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, mHandler.getBucket(), null)) {
          S3RestUtils.checkPathIsAlluxioDirectory(userFs, bucketPath, auditContext,
              mHandler.BUCKET_PATH_CACHE);

          LOG.debug("DeleteBucketTagging bucket={}", bucketPath);
          Map<String, ByteString> xattrMap = new HashMap<>();
          xattrMap.put(S3Constants.TAGGING_XATTR_KEY, ByteString.copyFrom(new byte[0]));
          SetAttributePOptions attrPOptions = SetAttributePOptions.newBuilder()
                  .putAllXattr(xattrMap)
                  .setXattrUpdateStrategy(File.XAttrUpdateStrategy.DELETE_KEYS)
                  .build();
          try {
            userFs.setAttribute(new AlluxioURI(bucketPath), attrPOptions);
          } catch (Exception e) {
            throw S3RestUtils.toBucketS3Exception(e, bucketPath, auditContext);
          }
          return Response.Status.NO_CONTENT;
        }
      });
    }
  } // end of DeleteBucketTaggingTask

  private static class DeleteBucketTask extends S3BucketTask {

    protected DeleteBucketTask(S3Handler handler, OpType opType) {
      super(handler, opType);
    }

    @Override
    public Response continueTask() {
      return S3RestUtils.call(mHandler.getBucket(), () -> {
        final String user = mHandler.getUser();
        final FileSystem userFs = S3RestUtils.createFileSystemForUser(user, mHandler.getMetaFS());
        String bucketPath = S3RestUtils.parsePath(AlluxioURI.SEPARATOR + mHandler.getBucket());

        try (S3AuditContext auditContext = mHandler.createAuditContext(
                mOPType.name(), user, mHandler.getBucket(), null)) {
          S3RestUtils.checkPathIsAlluxioDirectory(userFs, bucketPath, auditContext,
              mHandler.BUCKET_PATH_CACHE);
          // Delete the bucket.
          DeletePOptions options = DeletePOptions.newBuilder().setAlluxioOnly(Configuration
                          .get(PropertyKey.PROXY_S3_DELETE_TYPE)
                          .equals(Constants.S3_DELETE_IN_ALLUXIO_ONLY))
                  .build();
          try {
            userFs.delete(new AlluxioURI(bucketPath), options);
            mHandler.BUCKET_PATH_CACHE.put(bucketPath, false);
          } catch (Exception e) {
            throw S3RestUtils.toBucketS3Exception(e, bucketPath, auditContext);
          }
          return Response.Status.NO_CONTENT;
        }
      });
    }
  } // end of DeleteBucketTask
}
