public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region,
    final BulkLoadHFileRequest request, List<String> clusterIds) throws IOException {
    final List<Pair<byte[], String>> familyPaths = new ArrayList<>(request.getFamilyPathCount());
    for (ClientProtos.BulkLoadHFileRequest.FamilyPath el : request.getFamilyPathList()) {
      familyPaths.add(new Pair<>(el.getFamily().toByteArray(), el.getPath()));
    }

    Token<AuthenticationTokenIdentifier> userToken = null;
    if (userProvider.isHadoopSecurityEnabled()) {
      userToken = new Token<>(request.getFsToken().getIdentifier().toByteArray(),
        request.getFsToken().getPassword().toByteArray(), new Text(request.getFsToken().getKind()),
        new Text(request.getFsToken().getService()));
    }
    final String bulkToken = request.getBulkToken();
    User user = getActiveUser();
    final UserGroupInformation ugi = user.getUGI();
    if (userProvider.isHadoopSecurityEnabled()) {
      try {
        Token<AuthenticationTokenIdentifier> tok = ClientTokenUtil.obtainToken(conn).get();
        if (tok != null) {
          boolean b = ugi.addToken(tok);
          LOG.debug("token added " + tok + " for user " + ugi + " return=" + b);
        }
      } catch (Exception ioe) {
        LOG.warn("unable to add token", ioe);
      }
    }
    if (userToken != null) {
      ugi.addToken(userToken);
    } else if (userProvider.isHadoopSecurityEnabled()) {
      // we allow this to pass through in "simple" security mode
      // for mini cluster testing
      throw new DoNotRetryIOException("User token cannot be null");
    }

    if (region.getCoprocessorHost() != null) {
      region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
    }
    Map<byte[], List<Path>> map = null;

    try {
      incrementUgiReference(ugi);
      // Get the target fs (HBase region server fs) delegation token
      // Since we have checked the permission via 'preBulkLoadHFile', now let's give
      // the 'request user' necessary token to operate on the target fs.
      // After this point the 'doAs' user will hold two tokens, one for the source fs
      // ('request user'), another for the target fs (HBase region server principal).
      if (userProvider.isHadoopSecurityEnabled()) {
        FsDelegationToken targetfsDelegationToken = new FsDelegationToken(userProvider, "renewer");
        targetfsDelegationToken.acquireDelegationToken(fs);

        Token<?> targetFsToken = targetfsDelegationToken.getUserToken();
        if (
          targetFsToken != null
            && (userToken == null || !targetFsToken.getService().equals(userToken.getService()))
        ) {
          ugi.addToken(targetFsToken);
        }
      }

      map = ugi.doAs(new PrivilegedAction<Map<byte[], List<Path>>>() {
        @Override
        public Map<byte[], List<Path>> run() {
          FileSystem fs = null;
          try {
            /*
             * This is creating and caching a new FileSystem instance. Other code called "beneath"
             * this method will rely on this FileSystem instance being in the cache. This is
             * important as those methods make _no_ attempt to close this FileSystem instance. It is
             * critical that here, in SecureBulkLoadManager, we are tracking the lifecycle and
             * closing the FS when safe to do so.
             */
            fs = FileSystem.get(conf);
            for (Pair<byte[], String> el : familyPaths) {
              Path stageFamily = new Path(bulkToken, Bytes.toString(el.getFirst()));
              if (!fs.exists(stageFamily)) {
                fs.mkdirs(stageFamily);
                fs.setPermission(stageFamily, PERM_ALL_ACCESS);
              }
            }
            if (fsCreatedListener != null) {
              fsCreatedListener.accept(region);
            }
            // We call bulkLoadHFiles as requesting user
            // To enable access prior to staging
            return region.bulkLoadHFiles(familyPaths, true,
              new SecureBulkLoadListener(fs, bulkToken, conf), request.getCopyFile(), clusterIds,
              request.getReplicate());
          } catch (Exception e) {
            LOG.error("Failed to complete bulk load", e);
          }
          return null;
        }
      });
    } finally {
      decrementUgiReference(ugi);
      try {
        if (!UserGroupInformation.getLoginUser().equals(ugi) && !isUserReferenced(ugi)) {
          FileSystem.closeAllForUGI(ugi);
        }
      } catch (IOException e) {
        LOG.error("Failed to close FileSystem for: {}", ugi, e);
      }
      if (region.getCoprocessorHost() != null) {
        region.getCoprocessorHost().postBulkLoadHFile(familyPaths, map);
      }
    }
    return map;
  }
Bulk loads in secure mode. This service addresses two issues:
<ol>
<li>Moving files in a secure filesystem wherein the HBase Client and HBase Server are different
filesystem users.</li>
<li>Does moving in a secure manner. Assuming that the filesystem is POSIX compliant.</li>
</ol>
The algorithm is as follows:
<ol>
<li>Create an hbase owned staging directory which is world traversable (711):
{@code /hbase/staging}</li>
<li>A user writes out data to his secure output directory: {@code /user/foo/data}</li>
<li>A call is made to hbase to create a secret staging directory which globally rwx (777):
{@code /user/staging/averylongandrandomdirectoryname}</li>
<li>The user moves the data into the random staging directory, then calls bulkLoadHFiles()</li>
</ol>
Like delegation tokens the strength of the security lies in the length and randomness of the
secret directory.
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SecureBulkLoadManager.java