CDAP-21027 : upgrading hadoop version to 3.3.6

cdap-app-fabric-tests/pom.xml

@@ -95,12 +95,6 @@ the License.
       <artifactId>jcl-over-slf4j</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>io.cdap.cdap</groupId>
-      <artifactId>cdap-hbase-compat-1.0</artifactId>
-      <version>${project.version}</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>

cdap-app-fabric-tests/src/test/java/io/cdap/cdap/internal/app/runtime/batch/DynamicPartitionerWithAvroTest.java

@@ -172,6 +172,7 @@ public void apply() throws IOException {
     // run the partition writer m/r with this output partition time
     Map<String, String> arguments = new HashMap<>();
     arguments.put(OUTPUT_PARTITION_KEY, Long.toString(now));
+    arguments.put("system.mapreduce.mapreduce.fileoutputcommitter.algorithm.version", "1");
     arguments.put(allowConcurrencyKey, Boolean.toString(allowConcurrentWriters));
     if (partitionWriteOption != null) {
       arguments.put("partitionWriteOption", partitionWriteOption.name());

cdap-app-fabric/pom.xml

@@ -243,7 +243,7 @@
     </dependency>
     <dependency>
       <groupId>org.powermock</groupId>
-      <artifactId>powermock-api-mockito</artifactId>
+      <artifactId>powermock-api-mockito2</artifactId>
     </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
@@ -265,6 +265,11 @@
           <artifactId>cdap-event-reader-spi</artifactId>
           <version>${project.version}</version>
       </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-core</artifactId>
+      <version>2.2</version>
+    </dependency>
   </dependencies>
 
   <build>

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/batch/MapReduceProgramRunner.java

@@ -64,6 +64,7 @@
 import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import javax.annotation.Nullable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.JobContext;
@@ -81,6 +82,7 @@
 public class MapReduceProgramRunner extends AbstractProgramRunnerWithPlugin {
 
   private static final Logger LOG = LoggerFactory.getLogger(MapReduceProgramRunner.class);
+  public static final String MAPREDUCE_CUSTOM_CONFIG_PREFIX = "system.mapreduce.";
 
   private final Injector injector;
   private final CConfiguration cConf;
@@ -205,6 +207,9 @@ public ProgramController run(final Program program, ProgramOptions options) {
         hConf.set(JobContext.QUEUE_NAME, schedulerQueue);
       }
 
+      hConf = setCustomMapReduceConfig(hConf, options.getArguments());
+      hConf = setCustomMapReduceConfig(hConf, options.getUserArguments());
+
       ClusterMode clusterMode = ProgramRunners.getClusterMode(options);
       Service mapReduceRuntimeService = new MapReduceRuntimeService(injector, cConf, hConf,
           mapReduce, spec,
@@ -241,4 +246,19 @@ private File getPluginArchive(ProgramOptions options) {
     }
     return new File(options.getArguments().getOption(ProgramOptionConstants.PLUGIN_ARCHIVE));
   }
+
+  private Configuration setCustomMapReduceConfig(Configuration hConf, Arguments options) {
+    Map<String, String> systemArgs = options.asMap();
+    for (String name : systemArgs.keySet()) {
+      if (!name.startsWith(MAPREDUCE_CUSTOM_CONFIG_PREFIX)) {
+        continue;
+      }
+      String value = systemArgs.get(name);
+      String key = name.substring(MAPREDUCE_CUSTOM_CONFIG_PREFIX.length());
+      if (key != null && value != null) {
+        hConf.set(key, value);
+      }
+    }
+    return hConf;
+  }
 }

cdap-app-fabric/src/test/java/io/cdap/cdap/gateway/handlers/JsonListResponderTest.java

@@ -41,7 +41,7 @@ public void testSendForPaginatedListResponder() throws IOException {
     ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
 
     Mockito.doAnswer(invocation -> {
-      ByteBuffers.writeToStream(invocation.getArgumentAt(0, ByteBuffer.class), byteArrayOutputStream);
+      ByteBuffers.writeToStream(invocation.getArgument(0, ByteBuffer.class), byteArrayOutputStream);
       return null;
     }).when(chunkResponder).sendChunk(Mockito.any(ByteBuffer.class));
 
@@ -65,7 +65,7 @@ public void testMultipleSendForPaginatedListResponder() throws IOException {
     ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
 
     Mockito.doAnswer(invocation -> {
-      ByteBuffers.writeToStream(invocation.getArgumentAt(0, ByteBuffer.class), byteArrayOutputStream);
+      ByteBuffers.writeToStream(invocation.getArgument(0, ByteBuffer.class), byteArrayOutputStream);
       return null;
     }).when(chunkResponder).sendChunk(Mockito.any(ByteBuffer.class));
 
@@ -91,7 +91,7 @@ public void testSendForWholeListResponder() throws IOException {
     ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
 
     Mockito.doAnswer(invocation -> {
-      ByteBuffers.writeToStream(invocation.getArgumentAt(0, ByteBuffer.class), byteArrayOutputStream);
+      ByteBuffers.writeToStream(invocation.getArgument(0, ByteBuffer.class), byteArrayOutputStream);
       return null;
     }).when(chunkResponder).sendChunk(Mockito.any(ByteBuffer.class));
 

cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/runtime/RemoteAppStateStoreTest.java

@@ -36,22 +36,23 @@
 import io.cdap.cdap.proto.id.ApplicationId;
 import io.cdap.cdap.security.auth.context.AuthenticationTestContext;
 import io.cdap.http.NettyHttpService;
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Base64;
-import java.util.Optional;
 import org.apache.twill.common.Cancellable;
 import org.apache.twill.discovery.InMemoryDiscoveryService;
 import org.hamcrest.CoreMatchers;
-import org.hamcrest.CustomTypeSafeMatcher;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.mockito.ArgumentMatcher;
 import org.mockito.Mockito;
 
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Optional;
+
 /**
  * Tests for {@link RemoteAppStateStore}
  */
@@ -103,76 +104,70 @@ private static void setUpMockBehaviorForApplicationLifeCycleService() throws App
     //Throw ApplicationNotFoundException when ever NOT_FOUND_APP is used
     Mockito.doThrow(new ApplicationNotFoundException(new ApplicationId(NAMESPACE, NOT_FOUND_APP)))
       .when(applicationLifecycleService)
-      .saveState(Mockito.argThat(new AppNameAppStateKeyValueMatcher("NOT_FOUND_APP", NOT_FOUND_APP)));
+      .saveState(Mockito.argThat(new AppNameAppStateKeyValueMatcher(NOT_FOUND_APP)));
     Mockito.doThrow(new ApplicationNotFoundException(new ApplicationId(NAMESPACE, NOT_FOUND_APP)))
       .when(applicationLifecycleService)
-      .getState(Mockito.argThat(new AppNameAppStateKeyMatcher("NOT_FOUND_APP", NOT_FOUND_APP)));
+      .getState(Mockito.argThat(new AppNameAppStateKeyMatcher(NOT_FOUND_APP)));
     Mockito.doThrow(new ApplicationNotFoundException(new ApplicationId(NAMESPACE, NOT_FOUND_APP)))
         .when(applicationLifecycleService)
-        .deleteState(Mockito.argThat(new AppNameAppStateKeyMatcher("NOT_FOUND_APP", NOT_FOUND_APP)));
+        .deleteState(Mockito.argThat(new AppNameAppStateKeyMatcher( NOT_FOUND_APP)));
 
     //Throw RuntimeException whenever error app is being used
     Mockito.doThrow(new RuntimeException("test")).when(applicationLifecycleService)
-      .saveState(Mockito.argThat(new AppNameAppStateKeyValueMatcher("ERROR_APP", ERROR_APP)));
+      .saveState(Mockito.argThat(new AppNameAppStateKeyValueMatcher(ERROR_APP)));
     Mockito.doThrow(new RuntimeException("test")).when(applicationLifecycleService)
-      .getState(Mockito.argThat(new AppNameAppStateKeyMatcher("ERROR_APP", ERROR_APP)));
+      .getState(Mockito.argThat(new AppNameAppStateKeyMatcher(ERROR_APP)));
     Mockito.doThrow(new RuntimeException("test")).when(applicationLifecycleService)
-        .deleteState(Mockito.argThat(new AppNameAppStateKeyMatcher("ERROR_APP", ERROR_APP)));
+        .deleteState(Mockito.argThat(new AppNameAppStateKeyMatcher(ERROR_APP)));
 
     String encodedInvalidKey = Base64.getEncoder().encodeToString(MISSING_KEY.getBytes(StandardCharsets.UTF_8));
     // Different response for valid and invalid keys
     Mockito.when(
-        applicationLifecycleService.getState(
-          Mockito.argThat(new CustomTypeSafeMatcher<AppStateKey>("valid key match") {
-            @Override
-            protected boolean matchesSafely(AppStateKey item) {
-              return item.getAppName().equals(SUCCESS_APP) && !item.getStateKey().equals(encodedInvalidKey);
-            }
-          })))
+      applicationLifecycleService.getState(
+        Mockito.argThat(
+          item -> item == null ? false :
+            item.getAppName().equals(SUCCESS_APP) && !item.getStateKey().equals(encodedInvalidKey)
+        )))
       .thenReturn(Optional.of(TEST_VALUE.getBytes(StandardCharsets.UTF_8)));
 
     Mockito.when(
-        applicationLifecycleService.getState(
-          Mockito.argThat(new CustomTypeSafeMatcher<AppStateKey>("invalid key match") {
-            @Override
-            protected boolean matchesSafely(AppStateKey item) {
-              return item.getAppName().equals(SUCCESS_APP) && item.getStateKey().equals(encodedInvalidKey);
-            }
-          })))
+      applicationLifecycleService.getState(
+        Mockito.argThat(
+          item -> item == null ? false :
+            item.getAppName().equals(SUCCESS_APP) && item.getStateKey().equals(encodedInvalidKey)
+        )))
       .thenReturn(Optional.empty());
   }
 
   /**
    * Simple AppStateKeyValue matcher that matches for appname
    */
-  private static class AppNameAppStateKeyValueMatcher extends CustomTypeSafeMatcher<AppStateKeyValue> {
+  private static class AppNameAppStateKeyValueMatcher implements ArgumentMatcher<AppStateKeyValue> {
     private final String appName;
 
-    public AppNameAppStateKeyValueMatcher(String description, String appName) {
-      super(description);
+    public AppNameAppStateKeyValueMatcher(String appName) {
       this.appName = appName;
     }
 
     @Override
-    protected boolean matchesSafely(AppStateKeyValue item) {
-      return item.getAppName().equals(appName);
+    public boolean matches(AppStateKeyValue item) {
+      return item == null ? false : item.getAppName().equals(appName);
     }
   }
 
   /**
    * Simple AppStateKey matcher that matches for appname
    */
-  private static class AppNameAppStateKeyMatcher extends CustomTypeSafeMatcher<AppStateKey> {
+  private static class AppNameAppStateKeyMatcher implements ArgumentMatcher<AppStateKey> {
     private final String appName;
 
-    public AppNameAppStateKeyMatcher(String description, String appName) {
-      super(description);
+    public AppNameAppStateKeyMatcher(String appName) {
       this.appName = appName;
     }
 
     @Override
-    protected boolean matchesSafely(AppStateKey item) {
-      return item.getAppName().equals(appName);
+    public boolean matches(AppStateKey item) {
+      return item == null ? false : item.getAppName().equals(appName);
     }
   }
 

cdap-app-fabric/src/test/java/io/cdap/cdap/internal/capability/AutoInstallTest.java

@@ -152,7 +152,6 @@ public void testAutoInstallPlugins() throws Exception {
                                                    ImmutableSet.of());
     Mockito.verify(artifactRepository, Mockito.times(1)).writeArtifactProperties(artifact, properties);
     // Verify that temp file was deleted
-    PowerMockito.verifyStatic();
     java.nio.file.Files.deleteIfExists(mockPath);
   }