Merge branch 'trunk' of github.com:apache/hadoop into HADOOP-19218-trunk-2

Author: virajjasani

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ClosedIOException.java

@@ -0,0 +1,39 @@
+/**
+ * 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.fs;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Exception to denote if the underlying stream, cache or other closable resource
+ * is closed.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class ClosedIOException extends PathIOException {
+
+  /**
+   * Appends the custom error-message to the default error message.
+   * @param path path that encountered the closed resource.
+   * @param message custom error message.
+   */
+  public ClosedIOException(String path, String message) {
+    super(path, message);
+  }
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/HAServiceTarget.java

@@ -183,7 +183,7 @@ public final Map<String, String> getFencingParameters() {
    * expose to fencing implementations/scripts. Fencing methods are free
    * to use this map as they see fit -- notably, the shell script
    * implementation takes each entry, prepends 'target_', substitutes
-   * '_' for '.', and adds it to the environment of the script.
+   * '_' for '.' and '-', and adds it to the environment of the script.
    *
    * Subclass implementations should be sure to delegate to the superclass
    * implementation as well as adding their own keys.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ShellCommandFencer.java

@@ -39,8 +39,8 @@
  * (cmd.exe on Windows) and may not include any closing parentheses.<p>
  * 
  * The shell command will be run with an environment set up to contain
- * all of the current Hadoop configuration variables, with the '_' character 
- * replacing any '.' characters in the configuration keys.<p>
+ * all of the current Hadoop configuration variables, with the '_' character
+ * replacing any '.' or '-' characters in the configuration keys.<p>
  * 
  * If the shell command returns an exit code of 0, the fencing is
  * determined to be successful. If it returns any other exit code, the
@@ -202,11 +202,11 @@ private static String tryGetPid(Process p) {
 
   /**
    * Set the environment of the subprocess to be the Configuration,
-   * with '.'s replaced by '_'s.
+   * with '.'s and '-'s replaced by '_'s.
    */
   private void setConfAsEnvVars(Map<String, String> env) {
     for (Map.Entry<String, String> pair : getConf()) {
-      env.put(pair.getKey().replace('.', '_'), pair.getValue());
+      env.put(pair.getKey().replaceAll("[.-]", "_"), pair.getValue());
     }
   }
 
@@ -237,7 +237,7 @@ private void addTargetInfoAsEnvVars(HAServiceTarget target,
     for (Map.Entry<String, String> e :
          target.getFencingParameters().entrySet()) {
       String key = prefix + e.getKey();
-      key = key.replace('.', '_');
+      key = key.replaceAll("[.-]", "_");
       environment.put(key, e.getValue());
     }
   }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -106,6 +106,7 @@
 import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RPCTraceInfoProto;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.SaslConstants;
 import org.apache.hadoop.security.SaslPropertiesResolver;
 import org.apache.hadoop.security.SaslRpcServer;
 import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
@@ -2604,7 +2605,8 @@ private RpcSaslProto buildSaslNegotiateResponse()
       RpcSaslProto negotiateMessage = negotiateResponse;
       // accelerate token negotiation by sending initial challenge
       // in the negotiation response
-      if (enabledAuthMethods.contains(AuthMethod.TOKEN)) {
+      if (enabledAuthMethods.contains(AuthMethod.TOKEN)
+          && SaslConstants.SASL_MECHANISM_DEFAULT.equals(AuthMethod.TOKEN.getMechanismName())) {
         saslServer = createSaslServer(AuthMethod.TOKEN);
         byte[] challenge = saslServer.evaluateResponse(new byte[0]);
         RpcSaslProto.Builder negotiateBuilder =

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SaslConstants.java

@@ -32,7 +32,7 @@ public class SaslConstants {
 
   private static final String SASL_MECHANISM_ENV = "HADOOP_SASL_MECHANISM";
   public static final String SASL_MECHANISM;
-  private static final String SASL_MECHANISM_DEFAULT = "DIGEST-MD5";
+  public static final String SASL_MECHANISM_DEFAULT = "DIGEST-MD5";
 
   static {
     final String mechanism = System.getenv(SASL_MECHANISM_ENV);

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SaslRpcClient.java

@@ -39,6 +39,7 @@
 import javax.security.auth.callback.PasswordCallback;
 import javax.security.auth.callback.UnsupportedCallbackException;
 import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.sasl.AuthorizeCallback;
 import javax.security.sasl.RealmCallback;
 import javax.security.sasl.RealmChoiceCallback;
 import javax.security.sasl.Sasl;
@@ -681,9 +682,17 @@ public void handle(Callback[] callbacks)
           pc = (PasswordCallback) callback;
         } else if (callback instanceof RealmCallback) {
           rc = (RealmCallback) callback;
+        } else if (callback instanceof AuthorizeCallback) {
+          final AuthorizeCallback ac = (AuthorizeCallback) callback;
+          final String authId = ac.getAuthenticationID();
+          final String authzId = ac.getAuthorizationID();
+          ac.setAuthorized(authId.equals(authzId));
+          if (ac.isAuthorized()) {
+            ac.setAuthorizedID(authzId);
+          }
         } else {
           throw new UnsupportedCallbackException(callback,
-              "Unrecognized SASL client callback");
+              "Unrecognized SASL client callback " + callback.getClass());
         }
       }
       if (nc != null) {

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ha/TestShellCommandFencer.java

@@ -63,7 +63,7 @@ public void resetLogSpy() {
 
   private static ShellCommandFencer createFencer() {
     Configuration conf = new Configuration();
-    conf.set("in.fencing.tests", "yessir");
+    conf.set("in.fencing-tests", "yessir");
     ShellCommandFencer fencer = new ShellCommandFencer();
     fencer.setConf(conf);
     return fencer;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/HdfsClientConfigKeys.java

@@ -236,6 +236,9 @@ public interface HdfsClientConfigKeys {
   String DFS_DATA_TRANSFER_SASL_PROPS_RESOLVER_CLASS_KEY =
       "dfs.data.transfer.saslproperties.resolver.class";
 
+  String DFS_DATA_TRANSFER_SASL_CUSTOMIZEDCALLBACKHANDLER_CLASS_KEY
+      = "dfs.data.transfer.sasl.CustomizedCallbackHandler.class";
+
   String DFS_ENCRYPT_DATA_TRANSFER_CIPHER_KEY_BITLENGTH_KEY =
       "dfs.encrypt.data.transfer.cipher.key.bitlength";
   int    DFS_ENCRYPT_DATA_TRANSFER_CIPHER_KEY_BITLENGTH_DEFAULT = 128;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/DataTransferSaslUtil.java

@@ -102,9 +102,9 @@ public static void checkSaslComplete(SaslParticipant sasl,
     Set<String> requestedQop = ImmutableSet.copyOf(Arrays.asList(
         saslProps.get(Sasl.QOP).split(",")));
     String negotiatedQop = sasl.getNegotiatedQop();
-    LOG.debug("Verifying QOP, requested QOP = {}, negotiated QOP = {}",
-        requestedQop, negotiatedQop);
-    if (!requestedQop.contains(negotiatedQop)) {
+    LOG.debug("{}: Verifying QOP: requested = {}, negotiated = {}",
+        sasl, requestedQop, negotiatedQop);
+    if (negotiatedQop != null && !requestedQop.contains(negotiatedQop)) {
       throw new IOException(String.format("SASL handshake completed, but " +
           "channel does not have acceptable quality of protection, " +
           "requested = %s, negotiated = %s", requestedQop, negotiatedQop));

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/CustomizedCallbackHandler.java

@@ -0,0 +1,39 @@
+/*
+ * 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.hdfs.protocol.datatransfer.sasl;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import java.io.IOException;
+import java.util.List;
+
+/** For handling customized {@link Callback}. */
+public interface CustomizedCallbackHandler {
+  class DefaultHandler implements CustomizedCallbackHandler{
+    @Override
+    public void handleCallback(List<Callback> callbacks, String username, char[] password)
+        throws UnsupportedCallbackException {
+      if (!callbacks.isEmpty()) {
+        throw new UnsupportedCallbackException(callbacks.get(0));
+      }
+    }
+  }
+
+  void handleCallback(List<Callback> callbacks, String name, char[] password)
+      throws UnsupportedCallbackException, IOException;
+}