hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From te...@apache.org
Subject svn commit: r1433532 - in /hbase/branches/0.94: security/src/main/java/org/apache/hadoop/hbase/security/access/ security/src/test/java/org/apache/hadoop/hbase/mapreduce/ security/src/test/java/org/apache/hadoop/hbase/security/access/ src/main/java/org/...
Date Tue, 15 Jan 2013 17:15:16 GMT
Author: tedyu
Date: Tue Jan 15 17:15:16 2013
New Revision: 1433532

URL: http://svn.apache.org/viewvc?rev=1433532&view=rev
Log:
HBASE-5498 Secure Bulk Load (Francis Liu)


Added:
    hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java
    hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadProtocol.java
    hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/
    hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFiles.java
    hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFilesSplitRecovery.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/coprocessor/SecureBulkLoadClient.java
Modified:
    hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java
    hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/SecureTestUtil.java
    hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
    hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java
    hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java

Modified: hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java?rev=1433532&r1=1433531&r2=1433532&view=diff
==============================================================================
--- hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java (original)
+++ hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java Tue Jan 15 17:15:16 2013
@@ -18,7 +18,9 @@ import java.io.IOException;
 import java.net.InetAddress;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -65,6 +67,7 @@ import org.apache.hadoop.hbase.security.
 import org.apache.hadoop.hbase.security.access.Permission.Action;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.hadoop.hbase.util.Pair;
 
 import com.google.common.collect.ListMultimap;
 import com.google.common.collect.Lists;
@@ -452,7 +455,7 @@ public class AccessController extends Ba
    * @param families The map of column families-qualifiers.
    * @throws AccessDeniedException if the authorization check failed
    */
-  private void requirePermission(String request, Permission.Action perm,
+  public void requirePermission(String request, Permission.Action perm,
         RegionCoprocessorEnvironment env,
         Map<byte[], ? extends Collection<?>> families)
       throws IOException {
@@ -1017,6 +1020,72 @@ public class AccessController extends Ba
     }
   }
 
+  /**
+   * Verifies user has WRITE privileges on
+   * the Column Families involved in the bulkLoadHFile
+   * request. Specific Column Write privileges are presently
+   * ignored.
+   */
+  @Override
+  public void preBulkLoadHFile(ObserverContext<RegionCoprocessorEnvironment> ctx,
+      List<Pair<byte[], String>> familyPaths) throws IOException {
+    List<byte[]> cfs = new LinkedList<byte[]>();
+    for(Pair<byte[],String> el : familyPaths) {
+      cfs.add(el.getFirst());
+    }
+    requirePermission("bulkLoadHFile", Permission.Action.WRITE, ctx.getEnvironment(), cfs);
+  }
+
+  private AuthResult hasSomeAccess(RegionCoprocessorEnvironment e, String request, Action action) throws IOException {
+    User requestUser = getActiveUser();
+    byte[] tableName = e.getRegion().getTableDesc().getName();
+    AuthResult authResult = permissionGranted(request, requestUser,
+        action, e, Collections.EMPTY_MAP);
+    if (!authResult.isAllowed()) {
+      for(UserPermission userPerm:
+          AccessControlLists.getUserPermissions(regionEnv.getConfiguration(), tableName)) {
+        for(Permission.Action userAction: userPerm.getActions()) {
+          if(userAction.equals(action)) {
+            return AuthResult.allow(request, "Access allowed", requestUser,
+                action, tableName);
+          }
+        }
+      }
+    }
+    return authResult;
+  }
+
+  /**
+   * Authorization check for
+   * SecureBulkLoadProtocol.prepareBulkLoad()
+   * @param e
+   * @throws IOException
+   */
+  public void prePrepareBulkLoad(RegionCoprocessorEnvironment e) throws IOException {
+    AuthResult authResult = hasSomeAccess(e, "prepareBulkLoad", Action.WRITE);
+    logResult(authResult);
+    if (!authResult.isAllowed()) {
+      throw new AccessDeniedException("Insufficient permissions (table=" +
+        e.getRegion().getTableDesc().getNameAsString() + ", action=WRITE)");
+    }
+  }
+
+  /**
+   * Authorization security check for
+   * SecureBulkLoadProtocol.cleanupBulkLoad()
+   * @param e
+   * @throws IOException
+   */
+  //TODO this should end up as a coprocessor hook
+  public void preCleanupBulkLoad(RegionCoprocessorEnvironment e) throws IOException {
+    AuthResult authResult = hasSomeAccess(e, "cleanupBulkLoad", Action.WRITE);
+    logResult(authResult);
+    if (!authResult.isAllowed()) {
+      throw new AccessDeniedException("Insufficient permissions (table=" +
+        e.getRegion().getTableDesc().getNameAsString() + ", action=WRITE)");
+    }
+  }
+
   /* ---- AccessControllerProtocol implementation ---- */
   /*
    * These methods are only allowed to be called against the _acl_ region(s).

Added: hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java?rev=1433532&view=auto
==============================================================================
--- hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java (added)
+++ hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadEndpoint.java Tue Jan 15 17:15:16 2013
@@ -0,0 +1,317 @@
+/**
+ * Copyright 2011 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.security.access;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.coprocessor.BaseEndpointCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.ipc.RequestContext;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Methods;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+import java.math.BigInteger;
+import java.security.PrivilegedAction;
+import java.security.SecureRandom;
+import java.util.List;
+
+/**
+ * Coprocessor service for bulk loads in secure mode.
+ * This coprocessor has to be installed as part of enabling
+ * security in HBase.
+ *
+ * This service addresses two issues:
+ *
+ * 1. Moving files in a secure filesystem wherein the HBase Client
+ * and HBase Server are different filesystem users.
+ * 2. Does moving in a secure manner. Assuming that the filesystem
+ * is POSIX compliant.
+ *
+ * The algorithm is as follows:
+ *
+ * 1. Create an hbase owned staging directory which is
+ * world traversable (711): /hbase/staging
+ * 2. A user writes out data to his secure output directory: /user/foo/data
+ * 3. A call is made to hbase to create a secret staging directory
+ * which globally rwx (777): /user/staging/averylongandrandomdirectoryname
+ * 4. The user makes the data world readable and writable, then moves it
+ * into the random staging directory, then calls bulkLoadHFiles()
+ *
+ * Like delegation tokens the strength of the security lies in the length
+ * and randomness of the secret directory.
+ *
+ */
+@InterfaceAudience.Private
+public class SecureBulkLoadEndpoint extends BaseEndpointCoprocessor
+    implements SecureBulkLoadProtocol {
+
+  public static final long VERSION = 0L;
+
+  //Random number is 320 bits wide
+  private static final int RANDOM_WIDTH = 320;
+  //We picked 32 as the radix, so the character set
+  //will only contain alpha numeric values
+  //320/5 = 64 characters
+  private static final int RANDOM_RADIX = 32;
+
+  private static Log LOG = LogFactory.getLog(SecureBulkLoadEndpoint.class);
+
+  private final static FsPermission PERM_ALL_ACCESS = FsPermission.valueOf("-rwxrwxrwx");
+  private final static FsPermission PERM_HIDDEN = FsPermission.valueOf("-rwx--x--x");
+  private final static String BULKLOAD_STAGING_DIR = "hbase.bulkload.staging.dir";
+
+  private SecureRandom random;
+  private FileSystem fs;
+  private Configuration conf;
+
+  //two levels so it doesn't get deleted accidentally
+  //no sticky bit in Hadoop 1.0
+  private Path baseStagingDir;
+
+  private RegionCoprocessorEnvironment env;
+
+
+  @Override
+  public void start(CoprocessorEnvironment env) {
+    super.start(env);
+
+    this.env = (RegionCoprocessorEnvironment)env;
+    random = new SecureRandom();
+    conf = env.getConfiguration();
+    baseStagingDir = getBaseStagingDir(conf);
+
+    try {
+      fs = FileSystem.get(conf);
+      fs.mkdirs(baseStagingDir, PERM_HIDDEN);
+      fs.setPermission(baseStagingDir, PERM_HIDDEN);
+      //no sticky bit in hadoop-1.0, making directory nonempty so it never gets erased
+      fs.mkdirs(new Path(baseStagingDir,"DONOTERASE"), PERM_HIDDEN);
+      FileStatus status = fs.getFileStatus(baseStagingDir);
+      if(status == null) {
+        throw new IllegalStateException("Failed to create staging directory");
+      }
+      if(!status.getPermission().equals(PERM_HIDDEN)) {
+        throw new IllegalStateException("Directory already exists but permissions aren't set to '-rwx--x--x' ");
+      }
+    } catch (IOException e) {
+      throw new IllegalStateException("Failed to get FileSystem instance",e);
+    }
+  }
+
+  @Override
+  public String prepareBulkLoad(byte[] tableName) throws IOException {
+    getAccessController().prePrepareBulkLoad(env);
+    return createStagingDir(baseStagingDir, getActiveUser(), tableName).toString();
+  }
+
+  @Override
+  public void cleanupBulkLoad(String bulkToken) throws IOException {
+    getAccessController().preCleanupBulkLoad(env);
+    fs.delete(createStagingDir(baseStagingDir,
+        getActiveUser(),
+        env.getRegion().getTableDesc().getName(),
+        new Path(bulkToken).getName()),
+        true);
+  }
+
+  @Override
+  public boolean bulkLoadHFiles(final List<Pair<byte[], String>> familyPaths,
+                                final Token<?> userToken, final String bulkToken) throws IOException {
+    User user = getActiveUser();
+    final UserGroupInformation ugi = user.getUGI();
+    if(userToken != null) {
+      ugi.addToken(userToken);
+    } else if(User.isSecurityEnabled()) {
+      //we allow this to pass through in "simple" security mode
+      //for mini cluster testing
+      throw new DoNotRetryIOException("User token cannot be null");
+    }
+
+    HRegion region = env.getRegion();
+    boolean bypass = false;
+    if (region.getCoprocessorHost() != null) {
+      bypass = region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
+    }
+    boolean loaded = false;
+    if (!bypass) {
+      loaded = ugi.doAs(new PrivilegedAction<Boolean>() {
+        @Override
+        public Boolean run() {
+          FileSystem fs = null;
+          try {
+            Configuration conf = env.getConfiguration();
+            fs = FileSystem.get(conf);
+            for(Pair<byte[], String> el: familyPaths) {
+              Path p = new Path(el.getSecond());
+              LOG.debug("Setting permission for: " + p);
+              fs.setPermission(p, PERM_ALL_ACCESS);
+              Path stageFamily = new Path(bulkToken, Bytes.toString(el.getFirst()));
+              if(!fs.exists(stageFamily)) {
+                fs.mkdirs(stageFamily);
+                fs.setPermission(stageFamily, PERM_ALL_ACCESS);
+              }
+            }
+            //We call bulkLoadHFiles as requesting user
+            //To enable access prior to staging
+            return env.getRegion().bulkLoadHFiles(familyPaths,
+                new SecureBulkLoadListener(fs, bulkToken));
+          } catch (Exception e) {
+            LOG.error("Failed to complete bulk load", e);
+          }
+          return false;
+        }
+      });
+    }
+    if (region.getCoprocessorHost() != null) {
+      loaded = region.getCoprocessorHost().postBulkLoadHFile(familyPaths, loaded);
+    }
+    return loaded;
+  }
+
+  @Override
+  public long getProtocolVersion(String protocol, long clientVersion)
+      throws IOException {
+    if (SecureBulkLoadProtocol.class.getName().equals(protocol)) {
+      return SecureBulkLoadEndpoint.VERSION;
+    }
+    LOG.warn("Unknown protocol requested: " + protocol);
+    return -1;
+  }
+
+  private AccessController getAccessController() {
+    return (AccessController) this.env.getRegion()
+        .getCoprocessorHost().findCoprocessor(AccessController.class.getName());
+  }
+
+  private Path createStagingDir(Path baseDir, User user, byte[] tableName) throws IOException {
+    String randomDir = user.getShortName()+"__"+Bytes.toString(tableName)+"__"+
+        (new BigInteger(RANDOM_WIDTH, random).toString(RANDOM_RADIX));
+    return createStagingDir(baseDir, user, tableName, randomDir);
+  }
+
+  private Path createStagingDir(Path baseDir,
+                                User user,
+                                byte[] tableName,
+                                String randomDir) throws IOException {
+    Path p = new Path(baseDir, randomDir);
+    fs.mkdirs(p, PERM_ALL_ACCESS);
+    fs.setPermission(p, PERM_ALL_ACCESS);
+    return p;
+  }
+
+  private User getActiveUser() throws IOException {
+    User user = RequestContext.getRequestUser();
+    if (!RequestContext.isInRequestContext()) {
+      throw new DoNotRetryIOException("Failed to get requesting user");
+    }
+
+    //this is for testing
+    if("simple".equalsIgnoreCase(conf.get(User.HBASE_SECURITY_CONF_KEY))) {
+      return User.createUserForTesting(conf, user.getShortName(), new String[]{});
+    }
+
+    return user;
+  }
+
+  /**
+   * This returns the staging path for a given column family.
+   * This is needed for clean recovery and called reflectively in LoadIncrementalHFiles
+   */
+  public static Path getStagingPath(Configuration conf, String bulkToken, byte[] family) {
+    Path stageP = new Path(getBaseStagingDir(conf), bulkToken);
+    return new Path(stageP, Bytes.toString(family));
+  }
+
+  private static Path getBaseStagingDir(Configuration conf) {
+    return new Path(conf.get(BULKLOAD_STAGING_DIR, "/tmp/hbase-staging"));
+  }
+
+  private static class SecureBulkLoadListener implements HRegion.BulkLoadListener {
+    private FileSystem fs;
+    private String stagingDir;
+
+    public SecureBulkLoadListener(FileSystem fs, String stagingDir) {
+      this.fs = fs;
+      this.stagingDir = stagingDir;
+    }
+
+    @Override
+    public String prepareBulkLoad(final byte[] family, final String srcPath) throws IOException {
+      Path p = new Path(srcPath);
+      Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
+
+      if(!isFile(p)) {
+        throw new IOException("Path does not reference a file: " + p);
+      }
+
+      LOG.debug("Moving " + p + " to " + stageP);
+      if(!fs.rename(p, stageP)) {
+        throw new IOException("Failed to move HFile: " + p + " to " + stageP);
+      }
+      return stageP.toString();
+    }
+
+    @Override
+    public void doneBulkLoad(byte[] family, String srcPath) throws IOException {
+      LOG.debug("Bulk Load done for: " + srcPath);
+    }
+
+    @Override
+    public void failedBulkLoad(final byte[] family, final String srcPath) throws IOException {
+      Path p = new Path(srcPath);
+      Path stageP = new Path(stagingDir,
+          new Path(Bytes.toString(family), p.getName()));
+      LOG.debug("Moving " + stageP + " back to " + p);
+      if(!fs.rename(stageP, p))
+        throw new IOException("Failed to move HFile: " + stageP + " to " + p);
+    }
+
+    /**
+     * Check if the path is referencing a file.
+     * This is mainly needed to avoid symlinks.
+     * @param p
+     * @return true if the p is a file
+     * @throws IOException
+     */
+    private boolean isFile(Path p) throws IOException {
+      FileStatus status = fs.getFileStatus(p);
+      boolean isFile = !status.isDir();
+      try {
+        isFile = isFile && !(Boolean)Methods.call(FileStatus.class, status, "isSymlink", null, null);
+      } catch (Exception e) {
+      }
+      return isFile;
+    }
+  }
+}

Added: hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadProtocol.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadProtocol.java?rev=1433532&view=auto
==============================================================================
--- hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadProtocol.java (added)
+++ hbase/branches/0.94/security/src/main/java/org/apache/hadoop/hbase/security/access/SecureBulkLoadProtocol.java Tue Jan 15 17:15:16 2013
@@ -0,0 +1,67 @@
+/**
+ * Copyright 2011 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.security.access;
+
+import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
+import org.apache.hadoop.hbase.security.TokenInfo;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Provides a secure way to bulk load data onto HBase
+ * These are internal API. Bulk load should be initiated
+ * via {@link org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles}
+ * with security enabled.
+ */
+@TokenInfo("HBASE_AUTH_TOKEN")
+public interface SecureBulkLoadProtocol extends CoprocessorProtocol {
+
+  /**
+   * Prepare for bulk load.
+   * Will be called before bulkLoadHFiles()
+   * @param tableName
+   * @return a bulkToken which uniquely identifies the bulk session
+   * @throws IOException
+   */
+  String prepareBulkLoad(byte[] tableName) throws IOException;
+
+  /**
+   * Cleanup after bulk load.
+   * Will be called after bulkLoadHFiles().
+   * @param bulkToken
+   * @throws IOException
+   */
+  void cleanupBulkLoad(String bulkToken) throws IOException;
+
+  /**
+   * Secure version of HRegionServer.bulkLoadHFiles().
+   * @param familyPaths column family to HFile path pairs
+   * @param userToken requesting user's HDFS delegation token
+   * @param bulkToken
+   * @return
+   * @throws IOException
+   */
+  boolean bulkLoadHFiles(List<Pair<byte[], String>> familyPaths,
+                         Token<?> userToken, String bulkToken) throws IOException;
+
+}

Added: hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFiles.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFiles.java?rev=1433532&view=auto
==============================================================================
--- hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFiles.java (added)
+++ hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFiles.java Tue Jan 15 17:15:16 2013
@@ -0,0 +1,56 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.mapreduce;
+
+import org.apache.hadoop.hbase.LargeTests;
+import org.apache.hadoop.hbase.security.access.AccessControlLists;
+import org.apache.hadoop.hbase.security.access.SecureTestUtil;
+
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+
+/**
+ * Reruns TestLoadIncrementalHFiles using LoadIncrementalHFiles using secure mode.
+ * This suite is unable to verify the security handoff/turnover
+ * as miniCluster is running as system user thus has root privileges
+ * and delegation tokens don't seem to work on miniDFS.
+ *
+ * Thus SecureBulkload can only be completely verified by running
+ * integration tests against a secure cluster. This suite is still
+ * invaluable as it verifies the other mechanisms that need to be
+ * supported as part of a LoadIncrementalFiles call.
+ */
+@Category(LargeTests.class)
+public class TestSecureLoadIncrementalHFiles extends  TestLoadIncrementalHFiles{
+
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    useSecure = true;
+    // setup configuration
+    SecureTestUtil.enableSecurity(util.getConfiguration());
+
+    util.startMiniCluster();
+
+    // Wait for the ACL table to become available
+    util.waitTableAvailable(AccessControlLists.ACL_TABLE_NAME, 5000);
+  }
+
+}
+

Added: hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFilesSplitRecovery.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFilesSplitRecovery.java?rev=1433532&view=auto
==============================================================================
--- hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFilesSplitRecovery.java (added)
+++ hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/mapreduce/TestSecureLoadIncrementalHFilesSplitRecovery.java Tue Jan 15 17:15:16 2013
@@ -0,0 +1,66 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.mapreduce;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.LargeTests;
+import org.apache.hadoop.hbase.security.access.AccessControlLists;
+import org.apache.hadoop.hbase.security.access.SecureTestUtil;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+
+/**
+ * Reruns TestSecureLoadIncrementalHFilesSplitRecovery
+ * using LoadIncrementalHFiles in secure mode.
+ * This suite is unable to verify the security handoff/turnover
+ * as miniCluster is running as system user thus has root privileges
+ * and delegation tokens don't seem to work on miniDFS.
+ *
+ * Thus SecureBulkload can only be completely verified by running
+ * integration tests against a secure cluster. This suite is still
+ * invaluable as it verifies the other mechanisms that need to be
+ * supported as part of a LoadIncrementalFiles call.
+ */
+@Category(LargeTests.class)
+public class TestSecureLoadIncrementalHFilesSplitRecovery extends TestLoadIncrementalHFilesSplitRecovery {
+
+  //This "overrides" the parent static method
+  //make sure they are in sync
+  @BeforeClass
+  public static void setupCluster() throws Exception {
+    useSecure = true;
+    util = new HBaseTestingUtility();
+    // setup configuration
+    SecureTestUtil.enableSecurity(util.getConfiguration());
+
+    util.startMiniCluster();
+
+    // Wait for the ACL table to become available
+    util.waitTableAvailable(AccessControlLists.ACL_TABLE_NAME, 5000);
+  }
+
+  //Disabling this test as it does not work in secure mode
+  @Test
+  @Override
+  public void testBulkLoadPhaseFailure() {
+  }
+}
+

Modified: hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/SecureTestUtil.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/SecureTestUtil.java?rev=1433532&r1=1433531&r2=1433532&view=diff
==============================================================================
--- hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/SecureTestUtil.java (original)
+++ hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/SecureTestUtil.java Tue Jan 15 17:15:16 2013
@@ -34,7 +34,8 @@ public class SecureTestUtil {
     conf.set("hadoop.security.authentication", "simple");
     conf.set("hbase.rpc.engine", SecureRpcEngine.class.getName());
     conf.set(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY, AccessController.class.getName());
-    conf.set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY, AccessController.class.getName());
+    conf.set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY, AccessController.class.getName()+
+            ","+SecureBulkLoadEndpoint.class.getName());
     conf.set(CoprocessorHost.REGIONSERVER_COPROCESSOR_CONF_KEY, AccessController.class.getName());
     // add the process running user to superusers
     String currentUser = User.getCurrent().getName();

Modified: hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController.java?rev=1433532&r1=1433531&r2=1433532&view=diff
==============================================================================
--- hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController.java (original)
+++ hbase/branches/0.94/security/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController.java Tue Jan 15 17:15:16 2013
@@ -28,12 +28,17 @@ import java.util.List;
 import java.util.Map;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hbase.Coprocessor;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.HServerAddress;
 import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.LargeTests;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.UnknownRowLockException;
@@ -53,6 +58,9 @@ import org.apache.hadoop.hbase.coprocess
 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
 import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessorEnvironment;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.io.hfile.HFile;
+import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
 import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost;
@@ -192,22 +200,35 @@ public class TestAccessController {
       try {
         user.runAs(action);
         fail("Expected AccessDeniedException for user '" + user.getShortName() + "'");
-      } catch (RetriesExhaustedWithDetailsException e) {
-        // in case of batch operations, and put, the client assembles a
-        // RetriesExhaustedWithDetailsException instead of throwing an
-        // AccessDeniedException
+      } catch (AccessDeniedException ade) {
+        // expected result
+      } catch (IOException e) {
         boolean isAccessDeniedException = false;
-        for (Throwable ex : e.getCauses()) {
-          if (ex instanceof AccessDeniedException) {
-            isAccessDeniedException = true;
-            break;
+        if(e instanceof RetriesExhaustedWithDetailsException) {
+          // in case of batch operations, and put, the client assembles a
+          // RetriesExhaustedWithDetailsException instead of throwing an
+          // AccessDeniedException
+          for(Throwable ex : ((RetriesExhaustedWithDetailsException) e).getCauses()) {
+            if (ex instanceof AccessDeniedException) {
+              isAccessDeniedException = true;
+              break;
+            }
           }
         }
+        else {
+          // For doBulkLoad calls AccessDeniedException
+          // is buried in the stack trace
+          Throwable ex = e;
+          do {
+            if (ex instanceof AccessDeniedException) {
+              isAccessDeniedException = true;
+              break;
+            }
+          } while((ex = ex.getCause()) != null);
+        }
         if (!isAccessDeniedException) {
           fail("Not receiving AccessDeniedException for user '" + user.getShortName() + "'");
         }
-      } catch (AccessDeniedException ade) {
-        // expected result
       }
     }
   }
@@ -632,6 +653,104 @@ public class TestAccessController {
   }
 
   @Test
+  public void testBulkLoad() throws Exception {
+    FileSystem fs = TEST_UTIL.getTestFileSystem();
+    final Path dir = TEST_UTIL.getDataTestDir("testBulkLoad");
+    fs.mkdirs(dir);
+    //need to make it globally writable
+    //so users creating HFiles have write permissions
+    fs.setPermission(dir, FsPermission.valueOf("-rwxrwxrwx"));
+
+    PrivilegedExceptionAction bulkLoadAction = new PrivilegedExceptionAction() {
+      public Object run() throws Exception {
+        int numRows = 3;
+
+        //Making the assumption that the test table won't split between the range
+        byte[][][] hfileRanges = {{{(byte)0}, {(byte)9}}};
+
+        Path bulkLoadBasePath = new Path(dir, new Path(User.getCurrent().getName()));
+        new BulkLoadHelper(bulkLoadBasePath)
+            .bulkLoadHFile(TEST_TABLE, TEST_FAMILY, Bytes.toBytes("q"), hfileRanges, numRows);
+
+        return null;
+      }
+    };
+    verifyWrite(bulkLoadAction);
+  }
+
+  public class BulkLoadHelper {
+    private final FileSystem fs;
+    private final Path loadPath;
+    private final Configuration conf;
+
+    public BulkLoadHelper(Path loadPath) throws IOException {
+      fs = TEST_UTIL.getTestFileSystem();
+      conf = TEST_UTIL.getConfiguration();
+      loadPath = loadPath.makeQualified(fs);
+      this.loadPath = loadPath;
+    }
+
+    private void createHFile(Path path,
+        byte[] family, byte[] qualifier,
+        byte[] startKey, byte[] endKey, int numRows) throws IOException {
+
+      HFile.Writer writer = null;
+      long now = System.currentTimeMillis();
+      try {
+        writer = HFile.getWriterFactory(conf, new CacheConfig(conf))
+            .withPath(fs, path)
+            .withComparator(KeyValue.KEY_COMPARATOR)
+            .create();
+        // subtract 2 since numRows doesn't include boundary keys
+        for (byte[] key : Bytes.iterateOnSplits(startKey, endKey, true, numRows-2)) {
+          KeyValue kv = new KeyValue(key, family, qualifier, now, key);
+          writer.append(kv);
+        }
+      } finally {
+        if(writer != null)
+          writer.close();
+      }
+    }
+
+    private void bulkLoadHFile(
+        byte[] tableName,
+        byte[] family,
+        byte[] qualifier,
+        byte[][][] hfileRanges,
+        int numRowsPerRange) throws Exception {
+
+      Path familyDir = new Path(loadPath, Bytes.toString(family));
+      fs.mkdirs(familyDir);
+      int hfileIdx = 0;
+      for (byte[][] range : hfileRanges) {
+        byte[] from = range[0];
+        byte[] to = range[1];
+        createHFile(new Path(familyDir, "hfile_"+(hfileIdx++)),
+            family, qualifier, from, to, numRowsPerRange);
+      }
+      //set global read so RegionServer can move it
+      setPermission(loadPath, FsPermission.valueOf("-rwxrwxrwx"));
+
+      HTable table = new HTable(conf, tableName);
+      TEST_UTIL.waitTableAvailable(tableName, 30000);
+      LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
+      loader.doBulkLoad(loadPath, table);
+    }
+
+    public void setPermission(Path dir, FsPermission perm) throws IOException {
+      if(!fs.getFileStatus(dir).isDir()) {
+        fs.setPermission(dir,perm);
+      }
+      else {
+        for(FileStatus el : fs.listStatus(dir)) {
+          fs.setPermission(el.getPath(), perm);
+          setPermission(el.getPath() , perm);
+        }
+      }
+    }
+  }
+
+  @Test
   public void testAppend() throws Exception {
 
     PrivilegedExceptionAction appendAction = new PrivilegedExceptionAction() {

Added: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/coprocessor/SecureBulkLoadClient.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/coprocessor/SecureBulkLoadClient.java?rev=1433532&view=auto
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/coprocessor/SecureBulkLoadClient.java (added)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/coprocessor/SecureBulkLoadClient.java Tue Jan 15 17:15:16 2013
@@ -0,0 +1,92 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.coprocessor;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.util.Methods;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+import java.util.List;
+
+public class SecureBulkLoadClient {
+  private static Class protocolClazz;
+  private static Class endpointClazz;
+  private Object proxy;
+  private HTable table;
+
+  public SecureBulkLoadClient(HTable table) throws IOException {
+    this(table, HConstants.EMPTY_START_ROW);
+  }
+
+  public SecureBulkLoadClient(HTable table, byte[] startRow) throws IOException {
+    try {
+      protocolClazz = protocolClazz!=null?protocolClazz:
+          Class.forName("org.apache.hadoop.hbase.security.access.SecureBulkLoadProtocol");
+      endpointClazz = endpointClazz!=null?endpointClazz:
+          Class.forName("org.apache.hadoop.hbase.security.access.SecureBulkLoadEndpoint");
+      proxy = table.coprocessorProxy(protocolClazz, startRow);
+      this.table = table;
+    } catch (ClassNotFoundException e) {
+      throw new IOException("Failed to initialize SecureBulkLoad", e);
+    }
+  }
+
+  public String prepareBulkLoad(byte[] tableName) throws IOException {
+    try {
+      String bulkToken = (String) Methods.call(protocolClazz, proxy,
+          "prepareBulkLoad", new Class[]{byte[].class}, new Object[]{tableName});
+      return bulkToken;
+    } catch (Exception e) {
+      throw new IOException("Failed to prepareBulkLoad", e);
+    }
+  }
+
+  public void cleanupBulkLoad(String bulkToken) throws IOException {
+    try {
+      Methods.call(protocolClazz, proxy,
+          "cleanupBulkLoad", new Class[]{String.class},new Object[]{bulkToken});
+    } catch (Exception e) {
+      throw new IOException("Failed to prepareBulkLoad", e);
+    }
+  }
+
+  public boolean bulkLoadHFiles(List<Pair<byte[], String>> familyPaths,
+                         Token<?> userToken, String bulkToken) throws IOException {
+    try {
+      return (Boolean)Methods.call(protocolClazz, proxy, "bulkLoadHFiles",
+          new Class[]{List.class, Token.class, String.class},new Object[]{familyPaths, userToken, bulkToken});
+    } catch (Exception e) {
+      throw new IOException("Failed to bulkLoadHFiles", e);
+    }
+  }
+
+  public Path getStagingPath(String bulkToken, byte[] family) throws IOException {
+    try {
+      return (Path)Methods.call(endpointClazz, null, "getStagingPath",
+          new Class[]{Configuration.class, String.class, byte[].class},
+          new Object[]{table.getConfiguration(), bulkToken, family});
+    } catch (Exception e) {
+      throw new IOException("Failed to getStagingPath", e);
+    }
+  }
+}

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java?rev=1433532&r1=1433531&r2=1433532&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java Tue Jan 15 17:15:16 2013
@@ -60,6 +60,7 @@ import org.apache.hadoop.hbase.client.HB
 import org.apache.hadoop.hbase.client.HConnection;
 import org.apache.hadoop.hbase.client.HTable;
 import org.apache.hadoop.hbase.client.ServerCallable;
+import org.apache.hadoop.hbase.coprocessor.SecureBulkLoadClient;
 import org.apache.hadoop.hbase.io.HalfStoreFileReader;
 import org.apache.hadoop.hbase.io.Reference;
 import org.apache.hadoop.hbase.io.Reference.Range;
@@ -73,8 +74,10 @@ import org.apache.hadoop.hbase.io.hfile.
 import org.apache.hadoop.hbase.regionserver.Store;
 import org.apache.hadoop.hbase.regionserver.StoreFile;
 import org.apache.hadoop.hbase.regionserver.StoreFile.BloomType;
+import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
 
@@ -98,10 +101,21 @@ public class LoadIncrementalHFiles exten
 
   public static String NAME = "completebulkload";
 
-  public LoadIncrementalHFiles(Configuration conf) throws Exception {
+  private boolean useSecure;
+  private Token<?> userToken;
+  private String bulkToken;
+
+  //package private for testing
+  LoadIncrementalHFiles(Configuration conf, Boolean useSecure) throws Exception {
     super(conf);
     this.cfg = conf;
     this.hbAdmin = new HBaseAdmin(conf);
+    //added simple for testing
+    this.useSecure = useSecure != null ? useSecure : User.isHBaseSecurityEnabled(conf);
+  }
+
+  public LoadIncrementalHFiles(Configuration conf) throws Exception {
+    this(conf, null);
   }
 
   private void usage() {
@@ -212,6 +226,18 @@ public class LoadIncrementalHFiles exten
         return;
       }
 
+      //If using secure bulk load
+      //prepare staging directory and token
+      if(useSecure) {
+        //This condition is here for unit testing
+        //Since delegation token doesn't work in mini cluster
+        if(User.isSecurityEnabled()) {
+          FileSystem fs = FileSystem.get(cfg);
+          userToken = fs.getDelegationToken("renewer");
+        }
+        bulkToken = new SecureBulkLoadClient(table).prepareBulkLoad(table.getTableName());
+      }
+
       // Assumes that region splits can happen while this occurs.
       while (!queue.isEmpty()) {
         // need to reload split keys each iteration.
@@ -240,6 +266,18 @@ public class LoadIncrementalHFiles exten
       }
 
     } finally {
+      if(useSecure) {
+        if(userToken != null) {
+          try {
+            userToken.cancel(cfg);
+          } catch (Exception e) {
+            LOG.warn("Failed to cancel HDFS delegation token.", e);
+          }
+        }
+        if(bulkToken != null) {
+          new SecureBulkLoadClient(table).cleanupBulkLoad(bulkToken);
+        }
+      }
       pool.shutdown();
       if (queue != null && !queue.isEmpty()) {
         StringBuilder err = new StringBuilder();
@@ -473,11 +511,49 @@ public class LoadIncrementalHFiles exten
         tableName, first) {
       @Override
       public Boolean call() throws Exception {
-        LOG.debug("Going to connect to server " + location + " for row "
-            + Bytes.toStringBinary(row));
-        byte[] regionName = location.getRegionInfo().getRegionName();
-        return server.bulkLoadHFiles(famPaths, regionName);
+        SecureBulkLoadClient secureClient = null;
+        boolean success = false;
+
+        try {
+          LOG.debug("Going to connect to server " + location + " for row "
+              + Bytes.toStringBinary(row));
+          byte[] regionName = location.getRegionInfo().getRegionName();
+          if(!useSecure) {
+            success = server.bulkLoadHFiles(famPaths, regionName);
+          } else {
+            HTable table = new HTable(conn.getConfiguration(), tableName);
+            secureClient = new SecureBulkLoadClient(table, location.getRegionInfo().getStartKey());
+            success = secureClient.bulkLoadHFiles(famPaths, userToken, bulkToken);
+          }
+          return success;
+        } finally {
+          //Best effort copying of files that might not have been imported
+          //from the staging directory back to original location
+          //in user directory
+          if(secureClient != null && !success) {
+            FileSystem fs = FileSystem.get(cfg);
+            for(Pair<byte[], String> el : famPaths) {
+              Path hfileStagingPath = null;
+              Path hfileOrigPath = new Path(el.getSecond());
+              try {
+                hfileStagingPath= new Path(secureClient.getStagingPath(bulkToken, el.getFirst()),
+                    hfileOrigPath.getName());
+                if(fs.rename(hfileStagingPath, hfileOrigPath)) {
+                  LOG.debug("Moved back file " + hfileOrigPath + " from " +
+                      hfileStagingPath);
+                } else if(fs.exists(hfileStagingPath)){
+                  LOG.debug("Unable to move back file " + hfileOrigPath + " from " +
+                      hfileStagingPath);
+                }
+              } catch(Exception ex) {
+                LOG.debug("Unable to move back file " + hfileOrigPath + " from " +
+                    hfileStagingPath, ex);
+              }
+            }
+          }
+        }
       }
+
     };
 
     try {

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1433532&r1=1433531&r2=1433532&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Tue Jan 15 17:15:16 2013
@@ -3373,8 +3373,22 @@ public class HRegion implements HeapSize
    * @return true if successful, false if failed recoverably
    * @throws IOException if failed unrecoverably.
    */
-  public boolean bulkLoadHFiles(List<Pair<byte[], String>> familyPaths)
-  throws IOException {
+  public boolean bulkLoadHFiles(List<Pair<byte[], String>> familyPaths) throws IOException {
+    return bulkLoadHFiles(familyPaths, null);
+  }
+
+  /**
+   * Attempts to atomically load a group of hfiles.  This is critical for loading
+   * rows with multiple column families atomically.
+   *
+   * @param familyPaths List of Pair<byte[] column family, String hfilePath>
+   * @param bulkLoadListener Internal hooks enabling massaging/preparation of a
+   * file about to be bulk loaded
+   * @return true if successful, false if failed recoverably
+   * @throws IOException if failed unrecoverably.
+   */
+  public boolean bulkLoadHFiles(List<Pair<byte[], String>> familyPaths,
+      BulkLoadListener bulkLoadListener) throws IOException {
     Preconditions.checkNotNull(familyPaths);
     // we need writeLock for multi-family bulk load
     startBulkRegionOperation(hasMultipleColumnFamilies(familyPaths));
@@ -3434,7 +3448,14 @@ public class HRegion implements HeapSize
         String path = p.getSecond();
         Store store = getStore(familyName);
         try {
-          store.bulkLoadHFile(path);
+          String finalPath = path;
+          if(bulkLoadListener != null) {
+            finalPath = bulkLoadListener.prepareBulkLoad(familyName, path);
+          }
+          store.bulkLoadHFile(finalPath);
+          if(bulkLoadListener != null) {
+            bulkLoadListener.doneBulkLoad(familyName, path);
+          }
         } catch (IOException ioe) {
           // a failure here causes an atomicity violation that we currently
           // cannot recover from since it is likely a failed hdfs operation.
@@ -3442,6 +3463,14 @@ public class HRegion implements HeapSize
           // TODO Need a better story for reverting partial failures due to HDFS.
           LOG.error("There was a partial failure due to IO when attempting to" +
               " load " + Bytes.toString(p.getFirst()) + " : "+ p.getSecond());
+          if(bulkLoadListener != null) {
+            try {
+              bulkLoadListener.failedBulkLoad(familyName, path);
+            } catch (Exception ex) {
+              LOG.error("Error while calling failedBulkLoad for family "+
+                  Bytes.toString(familyName)+" with path "+path, ex);
+            }
+          }
           throw ioe;
         }
       }
@@ -5650,4 +5679,38 @@ public class HRegion implements HeapSize
        if (bc != null) bc.shutdown();
     }
   }
+
+  /**
+   * Listener class to enable callers of
+   * bulkLoadHFile() to perform any necessary
+   * pre/post processing of a given bulkload call
+   */
+  public static interface BulkLoadListener {
+
+    /**
+     * Called before an HFile is actually loaded
+     * @param family family being loaded to
+     * @param srcPath path of HFile
+     * @return final path to be used for actual loading
+     * @throws IOException
+     */
+    String prepareBulkLoad(byte[] family, String srcPath) throws IOException;
+
+    /**
+     * Called after a successful HFile load
+     * @param family family being loaded to
+     * @param srcPath path of HFile
+     * @throws IOException
+     */
+    void doneBulkLoad(byte[] family, String srcPath) throws IOException;
+
+    /**
+     * Called after a failed HFile load
+     * @param family family being loaded to
+     * @param srcPath path of HFile
+     * @throws IOException
+     */
+    void failedBulkLoad(byte[] family, String srcPath) throws IOException;
+
+  }
 }

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java?rev=1433532&r1=1433531&r2=1433532&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java Tue Jan 15 17:15:16 2013
@@ -577,7 +577,11 @@ public class Store extends SchemaConfigu
     // Move the file if it's on another filesystem
     FileSystem srcFs = srcPath.getFileSystem(conf);
     FileSystem desFs = fs instanceof HFileSystem ? ((HFileSystem)fs).getBackingFs() : fs;
-    if (!srcFs.equals(desFs)) {
+    //We can't compare FileSystem instances as
+    //equals() includes UGI instance as part of the comparison
+    //and won't work when doing SecureBulkLoad
+    //TODO deal with viewFS
+    if (!srcFs.getUri().equals(desFs.getUri())) {
       LOG.info("File " + srcPath + " on different filesystem than " +
           "destination store - moving to this filesystem.");
       Path tmpPath = getTmpPath();

Modified: hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java?rev=1433532&r1=1433531&r2=1433532&view=diff
==============================================================================
--- hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java (original)
+++ hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java Tue Jan 15 17:15:16 2013
@@ -60,7 +60,9 @@ public class TestLoadIncrementalHFiles {
   public static String COMPRESSION =
     Compression.Algorithm.NONE.getName();
 
-  private static HBaseTestingUtility util = new HBaseTestingUtility();
+  static HBaseTestingUtility util = new HBaseTestingUtility();
+  //used by secure subclass
+  static boolean useSecure = false;
 
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
@@ -149,8 +151,7 @@ public class TestLoadIncrementalHFiles {
 
     HTable table = new HTable(util.getConfiguration(), TABLE);
     util.waitTableAvailable(TABLE, 30000);
-    LoadIncrementalHFiles loader = new LoadIncrementalHFiles(
-      util.getConfiguration());
+    LoadIncrementalHFiles loader = new LoadIncrementalHFiles(util.getConfiguration(), useSecure);
     loader.doBulkLoad(dir, table);
 
     assertEquals(expectedRows, util.countRows(table));

Modified: hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java?rev=1433532&r1=1433531&r2=1433532&view=diff
==============================================================================
--- hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java (original)
+++ hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java Tue Jan 15 17:15:16 2013
@@ -68,7 +68,9 @@ import com.google.common.collect.Multima
 public class TestLoadIncrementalHFilesSplitRecovery {
   final static Log LOG = LogFactory.getLog(TestHRegionServerBulkLoad.class);
 
-  private static HBaseTestingUtility util;
+  static HBaseTestingUtility util;
+  //used by secure subclass
+  static boolean useSecure = false;
 
   final static int NUM_CFS = 10;
   final static byte[] QUAL = Bytes.toBytes("qual");
@@ -135,8 +137,7 @@ public class TestLoadIncrementalHFilesSp
    */
   private void populateTable(String table, int value) throws Exception {
     // create HFiles for different column families
-    LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
-        util.getConfiguration());
+    LoadIncrementalHFiles lih = new LoadIncrementalHFiles(util.getConfiguration(), useSecure);
     Path bulk1 = buildBulkFiles(table, value);
     HTable t = new HTable(util.getConfiguration(), Bytes.toBytes(table));
     lih.doBulkLoad(bulk1, t);
@@ -228,7 +229,7 @@ public class TestLoadIncrementalHFilesSp
     final AtomicInteger attmptedCalls = new AtomicInteger();
     final AtomicInteger failedCalls = new AtomicInteger();
     LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
-        util.getConfiguration()) {
+        util.getConfiguration(), useSecure) {
 
       protected List<LoadQueueItem> tryAtomicRegionLoad(final HConnection conn,
           byte[] tableName, final byte[] first, Collection<LoadQueueItem> lqis)
@@ -296,7 +297,7 @@ public class TestLoadIncrementalHFilesSp
     // files to fail when attempt to atomically import.  This is recoverable.
     final AtomicInteger attemptedCalls = new AtomicInteger();
     LoadIncrementalHFiles lih2 = new LoadIncrementalHFiles(
-        util.getConfiguration()) {
+        util.getConfiguration(), useSecure) {
 
       protected void bulkLoadPhase(final HTable htable, final HConnection conn,
           ExecutorService pool, Deque<LoadQueueItem> queue,
@@ -337,7 +338,7 @@ public class TestLoadIncrementalHFilesSp
 
     final AtomicInteger countedLqis= new AtomicInteger();
     LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
-        util.getConfiguration()) {
+        util.getConfiguration(), useSecure) {
       protected List<LoadQueueItem> groupOrSplit(
           Multimap<ByteBuffer, LoadQueueItem> regionGroups,
           final LoadQueueItem item, final HTable htable,
@@ -369,7 +370,7 @@ public class TestLoadIncrementalHFilesSp
     setupTable(table, 10);
 
     LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
-        util.getConfiguration()) {
+        util.getConfiguration(), useSecure) {
       int i = 0;
 
       protected List<LoadQueueItem> groupOrSplit(



Mime
View raw message