HDFS-17606. Do not require implementing CustomizedCallbackHandler.

Author: szetszwo

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

@@ -20,6 +20,8 @@
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.UnsupportedCallbackException;
 import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.util.List;
 
 /** For handling customized {@link Callback}. */
@@ -34,6 +36,24 @@ public void handleCallback(List<Callback> callbacks, String username, char[] pas
     }
   }
 
+  static CustomizedCallbackHandler delegate(Object delegated) {
+    return (callbacks, name, password) -> {
+      final String methodName = "handleCallback";
+      final Class<?> clazz = delegated.getClass();
+      final Method method;
+      try {
+        method = clazz.getMethod("handleCallback", List.class, String.class, char[].class);
+      } catch (NoSuchMethodException e) {
+        throw new IllegalStateException("Failed to get method " + methodName + " from " + clazz, e);
+      }
+      try {
+        method.invoke(delegated, callbacks, name, password);
+      } catch (IllegalAccessException | InvocationTargetException e) {
+        throw new IOException("Failed to invoke " + method, e);
+      }
+    };
+  }
+
   void handleCallback(List<Callback> callbacks, String name, char[] password)
       throws UnsupportedCallbackException, IOException;
 }

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

@@ -225,14 +225,20 @@ static final class SaslServerCallbackHandler
     SaslServerCallbackHandler(Configuration conf, PasswordFunction passwordFunction) {
       this.passwordFunction = passwordFunction;
 
-      final Class<? extends CustomizedCallbackHandler> clazz = conf.getClass(
+      final Class<?> clazz = conf.getClass(
           HdfsClientConfigKeys.DFS_DATA_TRANSFER_SASL_CUSTOMIZEDCALLBACKHANDLER_CLASS_KEY,
-          CustomizedCallbackHandler.DefaultHandler.class, CustomizedCallbackHandler.class);
+          CustomizedCallbackHandler.DefaultHandler.class);
+      final Object callbackHandler;
       try {
-        this.customizedCallbackHandler = clazz.newInstance();
+        callbackHandler = clazz.newInstance();
       } catch (Exception e) {
         throw new IllegalStateException("Failed to create a new instance of " + clazz, e);
       }
+      if (callbackHandler instanceof CustomizedCallbackHandler) {
+        customizedCallbackHandler = (CustomizedCallbackHandler) callbackHandler;
+      } else {
+        customizedCallbackHandler = CustomizedCallbackHandler.delegate(callbackHandler);
+      }
     }
 
     @Override

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/TestCustomizedCallbackHandler.java

@@ -60,4 +60,30 @@ public void testCustomizedCallbackHandler() throws Exception {
         MyCallbackHandler.class, CustomizedCallbackHandler.class);
     new SaslServerCallbackHandler(conf, String::toCharArray).handle(callbacks);
   }
+
+  static class MyCallbackMethod {
+    public void handleCallback(List<Callback> callbacks, String name, char[] password)
+        throws UnsupportedCallbackException {
+      LOG.info("{}: handling {} for {}", getClass().getSimpleName(), callbacks, name);
+    }
+  }
+
+  @Test
+  public void testCustomizedCallbackMethod() throws Exception {
+    final Configuration conf = new Configuration();
+    final Callback[] callbacks = {new MyCallback()};
+
+    // without setting conf, expect UnsupportedCallbackException
+    try {
+      new SaslServerCallbackHandler(conf, String::toCharArray).handle(callbacks);
+      Assert.fail("Expected UnsupportedCallbackException for " + Arrays.asList(callbacks));
+    } catch (UnsupportedCallbackException e) {
+      LOG.info("The failure is expected", e);
+    }
+
+    // set conf and expect success
+    conf.setClass(HdfsClientConfigKeys.DFS_DATA_TRANSFER_SASL_CUSTOMIZEDCALLBACKHANDLER_CLASS_KEY,
+        MyCallbackMethod.class, Object.class);
+    new SaslServerCallbackHandler(conf, String::toCharArray).handle(callbacks);
+  }
 }