void RGWCreateBucket::execute()
{
  RGWAccessControlPolicy old_policy(s->cct);
  buffer::list aclbl;
  buffer::list corsbl;
  bool existed;
  string bucket_name;
  rgw_make_bucket_entry_name(s->bucket_tenant, s->bucket_name, bucket_name);
  rgw_raw_obj obj(store->get_zone_params().domain_root, bucket_name);
  obj_version objv, *pobjv = NULL;

  op_ret = get_params();
  if (op_ret < 0)
    return;

  if (!location_constraint.empty() &&
      !store->has_zonegroup_api(location_constraint)) {
      ldout(s->cct, 0) << "location constraint (" << location_constraint << ")"
                       << " can't be found." << dendl;
      op_ret = -ERR_INVALID_LOCATION_CONSTRAINT;
      s->err.message = "The specified location-constraint is not valid";
      return;
  }

  if (!store->get_zonegroup().is_master_zonegroup() && !location_constraint.empty() &&
      store->get_zonegroup().api_name != location_constraint) {
    ldout(s->cct, 0) << "location constraint (" << location_constraint << ")"
                     << " doesn't match zonegroup" << " (" << store->get_zonegroup().api_name << ")"
                     << dendl;
    op_ret = -ERR_INVALID_LOCATION_CONSTRAINT;
    s->err.message = "The specified location-constraint is not valid";
    return;
  }

  const auto& zonegroup = store->get_zonegroup();
  if (!placement_rule.empty() &&
      !zonegroup.placement_targets.count(placement_rule)) {
    ldout(s->cct, 0) << "placement target (" << placement_rule << ")"
                     << " doesn't exist in the placement targets of zonegroup"
                     << " (" << store->get_zonegroup().api_name << ")" << dendl;
    op_ret = -ERR_INVALID_LOCATION_CONSTRAINT;
    s->err.message = "The specified placement target does not exist";
    return;
  }

  /* we need to make sure we read bucket info, it's not read before for this
   * specific request */
  RGWObjectCtx& obj_ctx = *static_cast<RGWObjectCtx *>(s->obj_ctx);
  op_ret = store->get_bucket_info(obj_ctx, s->bucket_tenant, s->bucket_name,
				  s->bucket_info, nullptr, &s->bucket_attrs);
  if (op_ret < 0 && op_ret != -ENOENT)
    return;
  s->bucket_exists = (op_ret != -ENOENT);

  s->bucket_owner.set_id(s->user->user_id);
  s->bucket_owner.set_name(s->user->display_name);
  if (s->bucket_exists) {
    int r = get_bucket_policy_from_attr(s->cct, store, s->bucket_info,
                                        s->bucket_attrs, &old_policy);
    if (r >= 0)  {
      if (old_policy.get_owner().get_id().compare(s->user->user_id) != 0) {
        op_ret = -EEXIST;
        return;
      }
    }
  }

  RGWBucketInfo master_info;
  rgw_bucket *pmaster_bucket;
  uint32_t *pmaster_num_shards;
  real_time creation_time;

  if (!store->is_meta_master()) {
    JSONParser jp;
    op_ret = forward_request_to_master(s, NULL, store, in_data, &jp);
    if (op_ret < 0) {
      return;
    }

    JSONDecoder::decode_json("entry_point_object_ver", ep_objv, &jp);
    JSONDecoder::decode_json("object_ver", objv, &jp);
    JSONDecoder::decode_json("bucket_info", master_info, &jp);
    ldout(s->cct, 20) << "parsed: objv.tag=" << objv.tag << " objv.ver=" << objv.ver << dendl;
    ldout(s->cct, 20) << "got creation time: << " << master_info.creation_time << dendl;
    pmaster_bucket= &master_info.bucket;
    creation_time = master_info.creation_time;
    pmaster_num_shards = &master_info.num_shards;
    pobjv = &objv;
  } else {
    pmaster_bucket = NULL;
    pmaster_num_shards = NULL;
  }

  string zonegroup_id;

  if (s->system_request) {
    zonegroup_id = s->info.args.get(RGW_SYS_PARAM_PREFIX "zonegroup");
    if (zonegroup_id.empty()) {
      zonegroup_id = store->get_zonegroup().get_id();
    }
  } else {
    zonegroup_id = store->get_zonegroup().get_id();
  }

  if (s->bucket_exists) {
    string selected_placement_rule;
    rgw_bucket bucket;
    bucket.tenant = s->bucket_tenant;
    bucket.name = s->bucket_name;
    op_ret = store->select_bucket_placement(*(s->user), zonegroup_id,
					    placement_rule,
					    &selected_placement_rule, nullptr);
    if (selected_placement_rule != s->bucket_info.placement_rule) {
      op_ret = -EEXIST;
      return;
    }
  }

  /* Encode special metadata first as we're using std::map::emplace under
   * the hood. This method will add the new items only if the map doesn't
   * contain such keys yet. */
  policy.encode(aclbl);
  emplace_attr(RGW_ATTR_ACL, std::move(aclbl));

  if (has_cors) {
    cors_config.encode(corsbl);
    emplace_attr(RGW_ATTR_CORS, std::move(corsbl));
  }

  RGWQuotaInfo quota_info;
  const RGWQuotaInfo * pquota_info = nullptr;
  if (need_metadata_upload()) {
    /* It's supposed that following functions WILL NOT change any special
     * attributes (like RGW_ATTR_ACL) if they are already present in attrs. */
    op_ret = rgw_get_request_metadata(s->cct, s->info, attrs, false);
    if (op_ret < 0) {
      return;
    }
    prepare_add_del_attrs(s->bucket_attrs, rmattr_names, attrs);
    populate_with_generic_attrs(s, attrs);

    op_ret = filter_out_quota_info(attrs, rmattr_names, quota_info);
    if (op_ret < 0) {
      return;
    } else {
      pquota_info = &quota_info;
    }

    /* Web site of Swift API. */
    filter_out_website(attrs, rmattr_names, s->bucket_info.website_conf);
    s->bucket_info.has_website = !s->bucket_info.website_conf.is_empty();
  }

  s->bucket.tenant = s->bucket_tenant; /* ignored if bucket exists */
  s->bucket.name = s->bucket_name;

  /* Handle updates of the metadata for Swift's object versioning. */
  if (swift_ver_location) {
    s->bucket_info.swift_ver_location = *swift_ver_location;
    s->bucket_info.swift_versioning = (! swift_ver_location->empty());
  }

  op_ret = store->create_bucket(*(s->user), s->bucket, zonegroup_id,
                                placement_rule, s->bucket_info.swift_ver_location,
                                pquota_info, attrs,
                                info, pobjv, &ep_objv, creation_time,
                                pmaster_bucket, pmaster_num_shards, true);
  /* continue if EEXIST and create_bucket will fail below.  this way we can
   * recover from a partial create by retrying it. */
  ldout(s->cct, 20) << "rgw_create_bucket returned ret=" << op_ret << " bucket=" << s->bucket << dendl;

  if (op_ret && op_ret != -EEXIST)
    return;

  existed = (op_ret == -EEXIST);

  if (existed) {
    /* bucket already existed, might have raced with another bucket creation, or
     * might be partial bucket creation that never completed. Read existing bucket
     * info, verify that the reported bucket owner is the current user.
     * If all is ok then update the user's list of buckets.
     * Otherwise inform client about a name conflict.
     */
    if (info.owner.compare(s->user->user_id) != 0) {
      op_ret = -EEXIST;
      return;
    }
    s->bucket = info.bucket;
  }

  op_ret = rgw_link_bucket(store, s->user->user_id, s->bucket,
			   info.creation_time, false);
  if (op_ret && !existed && op_ret != -EEXIST) {
    /* if it exists (or previously existed), don't remove it! */
    op_ret = rgw_unlink_bucket(store, s->user->user_id, s->bucket.tenant,
			       s->bucket.name);
    if (op_ret < 0) {
      ldout(s->cct, 0) << "WARNING: failed to unlink bucket: ret=" << op_ret
		       << dendl;
    }
  } else if (op_ret == -EEXIST || (op_ret == 0 && existed)) {
    op_ret = -ERR_BUCKET_EXISTS;
  }

  if (need_metadata_upload() && existed) {
    /* OK, it looks we lost race with another request. As it's required to
     * handle metadata fusion and upload, the whole operation becomes very
     * similar in nature to PutMetadataBucket. However, as the attrs may
     * changed in the meantime, we have to refresh. */
    short tries = 0;
    do {
      RGWObjectCtx& obj_ctx = *static_cast<RGWObjectCtx *>(s->obj_ctx);
      RGWBucketInfo binfo;
      map<string, bufferlist> battrs;

      op_ret = store->get_bucket_info(obj_ctx, s->bucket_tenant, s->bucket_name,
                                      binfo, nullptr, &battrs);
      if (op_ret < 0) {
        return;
      } else if (binfo.owner.compare(s->user->user_id) != 0) {
        /* New bucket doesn't belong to the account we're operating on. */
        op_ret = -EEXIST;
        return;
      } else {
        s->bucket_info = binfo;
        s->bucket_attrs = battrs;
      }

      attrs.clear();

      op_ret = rgw_get_request_metadata(s->cct, s->info, attrs, false);
      if (op_ret < 0) {
        return;
      }
      prepare_add_del_attrs(s->bucket_attrs, rmattr_names, attrs);
      populate_with_generic_attrs(s, attrs);
      op_ret = filter_out_quota_info(attrs, rmattr_names, s->bucket_info.quota);
      if (op_ret < 0) {
        return;
      }

      /* Handle updates of the metadata for Swift's object versioning. */
      if (swift_ver_location) {
        s->bucket_info.swift_ver_location = *swift_ver_location;
        s->bucket_info.swift_versioning = (! swift_ver_location->empty());
      }

      /* Web site of Swift API. */
      filter_out_website(attrs, rmattr_names, s->bucket_info.website_conf);
      s->bucket_info.has_website = !s->bucket_info.website_conf.is_empty();

      /* This will also set the quota on the bucket. */
      op_ret = rgw_bucket_set_attrs(store, s->bucket_info, attrs,
                                    &s->bucket_info.objv_tracker);
    } while (op_ret == -ECANCELED && tries++ < 20);

    /* Restore the proper return code. */
    if (op_ret >= 0) {
      op_ret = -ERR_BUCKET_EXISTS;
    }
  }
}