HIVE-28965: tez.DagUtils: Failed to add credential supplier, ClassNotFoundException: org.apache.hadoop.hive.kafka.KafkaDagCredentialSupplier (#6081)

Author: KiranVelumuri

ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java

@@ -45,6 +45,7 @@
 import java.util.Stack;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.zip.ZipOutputStream;
@@ -179,13 +180,13 @@ public class DagUtils {
   public static final String TEZ_TMP_DIR_KEY = "_hive_tez_tmp_dir";
   private static final Logger LOG = LoggerFactory.getLogger(DagUtils.class.getName());
   private static final String TEZ_DIR = "_tez_scratch_dir";
-  private static final DagUtils instance = new DagUtils(defaultCredentialSuppliers());
+  private static final DagUtils instance = new DagUtils(DagUtils::defaultCredentialSuppliers);
   // The merge file being currently processed.
   public static final String TEZ_MERGE_CURRENT_MERGE_FILE_PREFIX =
       "hive.tez.current.merge.file.prefix";
   // A comma separated list of work names used as prefix.
   public static final String TEZ_MERGE_WORK_FILE_PREFIXES = "hive.tez.merge.file.prefixes";
-  private final List<DagCredentialSupplier> credentialSuppliers;
+  private final Supplier<List<DagCredentialSupplier>> credentialSuppliers;
   /**
    * Notifiers to synchronize resource localization across threads. If one thread is localizing
    * a file, other threads can wait on the corresponding notifier object instead of just sleeping
@@ -286,7 +287,7 @@ private void getCredentialsFromSuppliers(BaseWork work, Set<TableDesc> tables, D
     if (!UserGroupInformation.isSecurityEnabled()){
       return;
     }
-    for (DagCredentialSupplier supplier : credentialSuppliers) {
+    for (DagCredentialSupplier supplier : credentialSuppliers.get()) {
       Text alias = supplier.getTokenAlias();
       Token<?> t = dag.getCredentials().getToken(alias);
       if (t != null) {
@@ -1697,7 +1698,7 @@ public static String getUserSpecifiedDagName(Configuration conf) {
   }
 
   @VisibleForTesting
-  DagUtils(List<DagCredentialSupplier> suppliers) {
+  DagUtils(Supplier<List<DagCredentialSupplier>> suppliers) {
     this.credentialSuppliers = suppliers;
   }
 

ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestDagUtilsSecurityEnabled.java

@@ -59,7 +59,7 @@ public static void clear() {
   @Test
   public void testAddCredentialsWithCredentialSupplierNewTokenAdded() {
     IncrementalIntDagCredentialSupplier supplier = new IncrementalIntDagCredentialSupplier();
-    DagUtils dagUtils = new DagUtils(Collections.singletonList(supplier));
+    DagUtils dagUtils = new DagUtils(() -> Collections.singletonList(supplier));
     DAG dag = DAG.create("test_credentials_dag");
 
     dagUtils.addCredentials(mock(MapWork.class), dag, null);
@@ -70,7 +70,7 @@ public void testAddCredentialsWithCredentialSupplierNewTokenAdded() {
   @Test
   public void testAddCredentialsWithCredentialSupplierTokenExistsNothingAdded() {
     IncrementalIntDagCredentialSupplier supplier = new IncrementalIntDagCredentialSupplier();
-    DagUtils dagUtils = new DagUtils(Collections.singletonList(supplier));
+    DagUtils dagUtils = new DagUtils(() -> Collections.singletonList(supplier));
     DAG dag = DAG.create("test_credentials_dag");
     Token<TokenIdentifier> oldToken = new Token<>();
     // Add explicitly the token in the DAG before calling addCredentials simulating the use-case where the DAG has already the token