TEZ-4311: Bump hadoop dependency version to 3.3.x

pom.xml

@@ -58,7 +58,7 @@
 
     <clover.license>${user.home}/clover.license</clover.license>
     <guava.version>27.0-jre</guava.version>
-    <hadoop.version>3.2.2</hadoop.version>
+    <hadoop.version>3.3.1</hadoop.version>
     <netty.version>4.1.61.Final</netty.version>
     <pig.version>0.13.0</pig.version>
     <jersey.version>1.19</jersey.version>
@@ -244,6 +244,12 @@
         <groupId>org.asynchttpclient</groupId>
         <artifactId>async-http-client</artifactId>
         <version>2.12.1</version>
+        <exclusions>
+          <exclusion>
+            <groupId>io.netty</groupId>
+            <artifactId>*</artifactId>
+          </exclusion>
+        </exclusions>
       </dependency>
       <dependency>
         <groupId>org.slf4j</groupId>
@@ -337,9 +343,10 @@
             <groupId>commons-el</groupId>
             <artifactId>commons-el</artifactId>
           </exclusion>
+          <!-- prevent dependency collision with hadoop-common => zookeeper => netty -->
           <exclusion>
             <groupId>io.netty</groupId>
-            <artifactId>netty</artifactId>
+            <artifactId>*</artifactId>
           </exclusion>
           <exclusion>
             <groupId>org.apache.httpcomponents</groupId>

tez-api/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClient.java

@@ -587,7 +587,11 @@ public void testTimelineClientCleanup() throws Exception {
     threadGroup.enumerate(threads);
     Thread reloaderThread = null;
     for (Thread thread : threads) {
-      if ((thread.getName() != null) && (thread.getName().contains("Truststore reloader thread"))) {
+      /* Since HADOOP-16524, the reloader thread's name is changed, let's handle the backward compatibility
+       * with a simple OR, as this is just a unit test, it's not worth involving a hadoop version check.
+       */
+      if ((thread.getName() != null) && (thread.getName().contains("Truststore reloader thread"))
+          || (thread.getName().contains("SSL Certificates Store Monitor"))) {
         reloaderThread = thread;
       }
     }

tez-plugins/tez-yarn-timeline-history-with-acls/src/test/java/org/apache/tez/dag/history/ats/acls/TestATSHistoryWithACLs.java

@@ -25,8 +25,10 @@
 import static org.mockito.Mockito.*;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Random;
 
 import javax.ws.rs.core.MediaType;
@@ -42,6 +44,7 @@
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineDomain;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.tez.client.TezClient;
 import org.apache.tez.common.ReflectionUtils;
@@ -63,6 +66,9 @@
 import org.apache.tez.runtime.library.processor.SleepProcessor;
 import org.apache.tez.runtime.library.processor.SleepProcessor.SleepProcessorConfig;
 import org.apache.tez.tests.MiniTezClusterWithTimeline;
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -149,9 +155,48 @@ private <K> K getTimelineData(String url, Class<K> clazz) {
     assertEquals(200, response.getStatus());
     assertTrue(MediaType.APPLICATION_JSON_TYPE.isCompatible(response.getType()));
 
-    K entity = response.getEntity(clazz);
-    assertNotNull(entity);
-    return entity;
+    JSONObject entity = response.getEntity(JSONObject.class);
+    K converted = null;
+    try {
+      converted = convertJSONObjectToTimelineObject(entity, clazz);
+    } catch (JSONException e) {
+      throw new RuntimeException(e);
+    }
+    assertNotNull(converted);
+    return converted;
+  }
+
+  private <K> K convertJSONObjectToTimelineObject(JSONObject jsonObj, Class<K> clazz) throws JSONException {
+    LOG.info("convertJSONObjectToEntity got object: " + jsonObj);
+    if (clazz == TimelineDomain.class) {
+      TimelineDomain domain = new TimelineDomain();
+      domain.setId(jsonObj.getString("id"));
+      domain.setOwner(jsonObj.getString("owner"));
+      domain.setReaders(jsonObj.getString("readers"));
+      domain.setWriters(jsonObj.getString("writers"));
+      return (K) domain;
+    } else if (clazz == TimelineEntity.class) {
+      TimelineEntity entity = new TimelineEntity();
+      entity.setEntityId(jsonObj.getString("entity"));
+      entity.setEntityType(jsonObj.getString("entitytype"));
+      entity.setDomainId(jsonObj.getString("domain"));
+      entity.setEvents(getEventsFromJSON(jsonObj));
+      return (K) entity;
+    } else {
+      throw new RuntimeException(
+          "convertJSONObjectToTimelineObject doesn't support conversion from JSONObject to " + clazz);
+    }
+  }
+
+  private List<TimelineEvent> getEventsFromJSON(JSONObject jsonObj) throws JSONException {
+    List<TimelineEvent> events = new ArrayList<>();
+    JSONArray arrEvents = jsonObj.getJSONArray("events");
+    for (int i = 0; i < arrEvents.length(); i++) {
+      TimelineEvent event = new TimelineEvent();
+      event.setEventType(((JSONObject) arrEvents.get(i)).getString("eventtype"));
+      events.add(event);
+    }
+    return events;
   }
 
   private TimelineDomain getDomain(String domainId) {
@@ -459,7 +504,8 @@ public void testDagLoggingEnabled() throws Exception {
         .get(ClientResponse.class);
     assertEquals(200, response.getStatus());
     assertTrue(MediaType.APPLICATION_JSON_TYPE.isCompatible(response.getType()));
-    TimelineEntity entity = response.getEntity(TimelineEntity.class);
+    JSONObject entityJson = response.getEntity(JSONObject.class);
+    TimelineEntity entity = convertJSONObjectToTimelineObject(entityJson, TimelineEntity.class);
     assertEquals(entity.getEntityType(), "TEZ_DAG_ID");
     assertEquals(entity.getEvents().get(0).getEventType(), HistoryEventType.DAG_SUBMITTED.toString());
   }

tez-runtime-library/pom.xml

@@ -33,6 +33,10 @@
       <groupId>org.asynchttpclient</groupId>
       <artifactId>async-http-client</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-all</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.tez</groupId>
       <artifactId>tez-api</artifactId>

tez-tests/src/test/java/org/apache/tez/test/TestSecureShuffle.java

@@ -114,6 +114,7 @@ public static Collection<Object[]> getParameters() {
   public static void setupDFSCluster() throws Exception {
     conf = new Configuration();
     conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_EDITS_NOEDITLOGCHANNELFLUSH, false);
+    conf.setBoolean("fs.hdfs.impl.disable.cache", true);
     EditLogFileOutputStream.setShouldSkipFsyncForTesting(true);
     conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
     miniDFSCluster =
@@ -301,8 +302,8 @@ public static X509Certificate generateCertificate(String dn, KeyPair pair, int d
 
     String hostAddress = InetAddress.getLocalHost().getHostAddress();
     certGen.addExtension(X509Extensions.SubjectAlternativeName, false,
-        new GeneralNames(new GeneralName(GeneralName.iPAddress, hostAddress)));
-
+        new GeneralNames(new GeneralName[] { new GeneralName(GeneralName.iPAddress, hostAddress),
+            new GeneralName(GeneralName.dNSName, "localhost") }));
     X500Principal dnName = new X500Principal(dn);
 
     certGen.setSerialNumber(sn);