int RGWBulkUploadOp::handle_dir(const boost::string_ref path)
{
  ldout(s->cct, 20) << "bulk upload: got directory=" << path << dendl;

  op_ret = handle_dir_verify_permission();
  if (op_ret < 0) {
    return op_ret;
  }

  std::string bucket_name;
  rgw_obj_key object_junk;
  std::tie(bucket_name, object_junk) =  *parse_path(path);

  rgw_raw_obj obj(store->get_zone_params().domain_root,
                  rgw_make_bucket_entry_name(s->bucket_tenant, bucket_name));

  /* we need to make sure we read bucket info, it's not read before for this
   * specific request */
  RGWBucketInfo binfo;
  std::map<std::string, ceph::bufferlist> battrs;
  op_ret = store->get_bucket_info(*dir_ctx, s->bucket_tenant, bucket_name,
                                  binfo, nullptr, &battrs);
  if (op_ret < 0 && op_ret != -ENOENT) {
    return op_ret;
  }
  const bool bucket_exists = (op_ret != -ENOENT);

  if (bucket_exists) {
    RGWAccessControlPolicy old_policy(s->cct);
    int r = get_bucket_policy_from_attr(s->cct, store, binfo,
                                        battrs, &old_policy);
    if (r >= 0)  {
      if (old_policy.get_owner().get_id().compare(s->user->user_id) != 0) {
        op_ret = -EEXIST;
        return op_ret;
      }
    }
  }

  RGWBucketInfo master_info;
  rgw_bucket *pmaster_bucket = nullptr;
  uint32_t *pmaster_num_shards = nullptr;
  real_time creation_time;
  obj_version objv, ep_objv, *pobjv = nullptr;

  if (! store->is_meta_master()) {
    JSONParser jp;
    ceph::bufferlist in_data;
    req_info info = s->info;
    forward_req_info(s->cct, info, bucket_name);
    op_ret = forward_request_to_master(s, nullptr, store, in_data, &jp, &info);
    if (op_ret < 0) {
      return op_ret;
    }

    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 = nullptr;
    pmaster_num_shards = nullptr;
  }


  std::string placement_rule;
  if (bucket_exists) {
    std::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),
                                            store->get_zonegroup().get_id(),
                                            placement_rule,
                                            &selected_placement_rule,
                                            nullptr);
    if (selected_placement_rule != binfo.placement_rule) {
      op_ret = -EEXIST;
      ldout(s->cct, 20) << "bulk upload: non-coherent placement rule" << dendl;
      return op_ret;
    }
  }

  /* Create metadata: ACLs. */
  std::map<std::string, ceph::bufferlist> attrs;
  RGWAccessControlPolicy policy;
  policy.create_default(s->user->user_id, s->user->display_name);
  ceph::bufferlist aclbl;
  policy.encode(aclbl);
  attrs.emplace(RGW_ATTR_ACL, std::move(aclbl));

  RGWQuotaInfo quota_info;
  const RGWQuotaInfo * pquota_info = nullptr;

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


  RGWBucketInfo out_info;
  op_ret = store->create_bucket(*(s->user),
                                bucket,
                                store->get_zonegroup().get_id(),
                                placement_rule, binfo.swift_ver_location,
                                pquota_info, attrs,
                                out_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=" << bucket << dendl;

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

  const bool 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 (out_info.owner.compare(s->user->user_id) != 0) {
      op_ret = -EEXIST;
      ldout(s->cct, 20) << "bulk upload: conflicting bucket name" << dendl;
      return op_ret;
    }
    bucket = out_info.bucket;
  }

  op_ret = rgw_link_bucket(store, s->user->user_id, bucket,
                           out_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,
                               bucket.tenant, bucket.name);
    if (op_ret < 0) {
      ldout(s->cct, 0) << "bulk upload: WARNING: failed to unlink bucket: ret="
                       << op_ret << dendl;
    }
  } else if (op_ret == -EEXIST || (op_ret == 0 && existed)) {
    ldout(s->cct, 20) << "bulk upload: containers already exists"
                      << dendl;
    op_ret = -ERR_BUCKET_EXISTS;
  }

  return op_ret;
}