merge with trunk

hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh

@@ -1571,12 +1571,21 @@ function hadoop_finalize_hadoop_opts
 
 ## @description  Finish configuring JPMS that enforced for JDK 17 and higher
 ## @description  prior to executing Java
+## @description  keep this list sync with hadoop-project/pom.xml extraJavaTestArgs
 ## @audience     private
 ## @stability    evolving
 ## @replaceable  yes
 function hadoop_finalize_jpms_opts
 {
     hadoop_add_param HADOOP_OPTS IgnoreUnrecognizedVMOptions "-XX:+IgnoreUnrecognizedVMOptions"
+    hadoop_add_param HADOOP_OPTS open.java.io "--add-opens=java.base/java.io=ALL-UNNAMED"
+    hadoop_add_param HADOOP_OPTS open.java.lang "--add-opens=java.base/java.lang=ALL-UNNAMED"
+    hadoop_add_param HADOOP_OPTS open.java.lang.reflect "--add-opens=java.base/java.lang.reflect=ALL-UNNAMED"
+    hadoop_add_param HADOOP_OPTS open.java.math "--add-opens=java.base/java.math=ALL-UNNAMED"
+    hadoop_add_param HADOOP_OPTS open.java.net "--add-opens=java.base/java.net=ALL-UNNAMED"
+    hadoop_add_param HADOOP_OPTS open.java.text "--add-opens=java.base/java.text=ALL-UNNAMED"
+    hadoop_add_param HADOOP_OPTS open.java.util "--add-opens=java.base/java.util=ALL-UNNAMED"
+    hadoop_add_param HADOOP_OPTS open.java.util.concurrent "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED"
     hadoop_add_param HADOOP_OPTS open.java.util.zip "--add-opens=java.base/java.util.zip=ALL-UNNAMED"
     hadoop_add_param HADOOP_OPTS open.sun.security.util "--add-opens=java.base/sun.security.util=ALL-UNNAMED"
     hadoop_add_param HADOOP_OPTS open.sun.security.x509 "--add-opens=java.base/sun.security.x509=ALL-UNNAMED"

hadoop-common-project/hadoop-registry/pom.xml

@@ -233,7 +233,7 @@
       <configuration>
         <reuseForks>false</reuseForks>
         <forkedProcessTimeoutInSeconds>900</forkedProcessTimeoutInSeconds>
-        <argLine>-Xmx1024m -XX:+HeapDumpOnOutOfMemoryError</argLine>
+        <argLine>${maven-surefire-plugin.argLine} -Xmx1024m -XX:+HeapDumpOnOutOfMemoryError</argLine>
         <environmentVariables>
           <!-- HADOOP_HOME required for tests on Windows to find winutils -->
           <HADOOP_HOME>${hadoop.common.build.dir}</HADOOP_HOME>

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/RedundantEditLogInputStream.java

@@ -193,6 +193,8 @@ protected FSEditLogOp nextOp() throws IOException {
         } catch (IOException e) {
           prevException = e;
           state = State.STREAM_FAILED;
+          LOG.warn("Got error skipUntil edit log input stream {}.", streams[curIdx].getName());
+          break;
         }
         state = State.OK;
         break;

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestRedundantEditLogInputStream.java

@@ -0,0 +1,67 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.server.namenode;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import org.junit.Test;
+
+import org.apache.hadoop.hdfs.server.namenode.FSEditLogOp.MkdirOp;
+import org.apache.hadoop.test.GenericTestUtils.LogCapturer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestRedundantEditLogInputStream {
+  private static final String FAKE_EDIT_STREAM_NAME = "FAKE_STREAM";
+
+  @Test
+  public void testNextOp() throws IOException {
+    EditLogInputStream fakeStream1 = mock(EditLogInputStream.class);
+    EditLogInputStream fakeStream2 = mock(EditLogInputStream.class);
+    ArrayList<EditLogInputStream> list = new ArrayList();
+    list.add(fakeStream1);
+    list.add(fakeStream2);
+    for (int i = 0; i < list.size(); i++) {
+      EditLogInputStream stream = list.get(i);
+      when(stream.getName()).thenReturn(FAKE_EDIT_STREAM_NAME + i);
+      when(stream.getFirstTxId()).thenReturn(1L);
+      when(stream.getLastTxId()).thenReturn(2L);
+      when(stream.length()).thenReturn(1L);
+    }
+    when(fakeStream1.skipUntil(1)).thenThrow(new IOException("skipUntil failed."));
+    when(fakeStream2.skipUntil(1)).thenReturn(true);
+    FSEditLogOp op = new MkdirOp();
+    op.setTransactionId(100);
+    when(fakeStream2.readOp()).thenReturn(op);
+
+    LogCapturer capture = LogCapturer.captureLogs(RedundantEditLogInputStream.LOG);
+    RedundantEditLogInputStream redundantEditLogInputStream =
+        new RedundantEditLogInputStream(list, 1);
+
+    FSEditLogOp returnOp = redundantEditLogInputStream.nextOp();
+    String log = capture.getOutput();
+    assertTrue(log.contains("Got error skipUntil edit log input stream FAKE_STREAM0"));
+    assertTrue(log.contains("Got error reading edit log input stream FAKE_STREAM0; "
+        + "failing over to edit log FAKE_STREAM1"));
+    assertEquals(op, returnOp);
+  }
+}

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/LineRecordReader.java

@@ -302,6 +302,8 @@ public synchronized void close() throws IOException {
     try {
       if (in != null) {
         in.close();
+      } else if (fileIn != null) {
+        fileIn.close();
       }
     } finally {
       if (decompressor != null) {

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/LineRecordReader.java

@@ -98,48 +98,53 @@ public void initialize(InputSplit genericSplit,
         MRJobConfig.INPUT_FILE_OPTION_PREFIX,
         MRJobConfig.INPUT_FILE_MANDATORY_PREFIX);
     fileIn = FutureIO.awaitFuture(builder.build());
-
-    CompressionCodec codec = new CompressionCodecFactory(job).getCodec(file);
-    if (null!=codec) {
-      isCompressedInput = true;
-      decompressor = CodecPool.getDecompressor(codec);
-      if (codec instanceof SplittableCompressionCodec) {
-        final SplitCompressionInputStream cIn =
-          ((SplittableCompressionCodec)codec).createInputStream(
-            fileIn, decompressor, start, end,
-            SplittableCompressionCodec.READ_MODE.BYBLOCK);
-        in = new CompressedSplitLineReader(cIn, job,
-            this.recordDelimiterBytes);
-        start = cIn.getAdjustedStart();
-        end = cIn.getAdjustedEnd();
-        filePosition = cIn;
-      } else {
-        if (start != 0) {
-          // So we have a split that is only part of a file stored using
-          // a Compression codec that cannot be split.
-          throw new IOException("Cannot seek in " +
-              codec.getClass().getSimpleName() + " compressed stream");
-        }
 
-        in = new SplitLineReader(codec.createInputStream(fileIn,
-            decompressor), job, this.recordDelimiterBytes);
+    try {
+      CompressionCodec codec = new CompressionCodecFactory(job).getCodec(file);
+      if (null!=codec) {
+        isCompressedInput = true;
+        decompressor = CodecPool.getDecompressor(codec);
+        if (codec instanceof SplittableCompressionCodec) {
+          final SplitCompressionInputStream cIn =
+                  ((SplittableCompressionCodec)codec).createInputStream(
+                          fileIn, decompressor, start, end,
+                          SplittableCompressionCodec.READ_MODE.BYBLOCK);
+          in = new CompressedSplitLineReader(cIn, job,
+                  this.recordDelimiterBytes);
+          start = cIn.getAdjustedStart();
+          end = cIn.getAdjustedEnd();
+          filePosition = cIn;
+        } else {
+          if (start != 0) {
+            // So we have a split that is only part of a file stored using
+            // a Compression codec that cannot be split.
+            throw new IOException("Cannot seek in " +
+                    codec.getClass().getSimpleName() + " compressed stream");
+          }
+
+          in = new SplitLineReader(codec.createInputStream(fileIn,
+                  decompressor), job, this.recordDelimiterBytes);
+          filePosition = fileIn;
+        }
+      } else {
+        fileIn.seek(start);
+        in = new UncompressedSplitLineReader(
+                fileIn, job, this.recordDelimiterBytes, split.getLength());
         filePosition = fileIn;
       }
-    } else {
-      fileIn.seek(start);
-      in = new UncompressedSplitLineReader(
-          fileIn, job, this.recordDelimiterBytes, split.getLength());
-      filePosition = fileIn;
-    }
-    // If this is not the first split, we always throw away first record
-    // because we always (except the last split) read one extra line in
-    // next() method.
-    if (start != 0) {
-      start += in.readLine(new Text(), 0, maxBytesToConsume(start));
+      // If this is not the first split, we always throw away first record
+      // because we always (except the last split) read one extra line in
+      // next() method.
+      if (start != 0) {
+        start += in.readLine(new Text(), 0, maxBytesToConsume(start));
+      }
+      this.pos = start;
+    } catch (Exception e) {
+      fileIn.close();
+      throw e;
     }
-    this.pos = start;
   }
-  
+
 
   private int maxBytesToConsume(long pos) {
     return isCompressedInput

hadoop-project/pom.xml

@@ -167,8 +167,19 @@
     <enforced.java.version>[${javac.version},)</enforced.java.version>
     <enforced.maven.version>[3.3.0,)</enforced.maven.version>
 
+    <!-- keep this list sync with
+         hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh#hadoop_finalize_jpms_opts
+    -->
     <extraJavaTestArgs>
       -XX:+IgnoreUnrecognizedVMOptions
+      --add-opens=java.base/java.io=ALL-UNNAMED
+      --add-opens=java.base/java.lang=ALL-UNNAMED
+      --add-opens=java.base/java.lang.reflect=ALL-UNNAMED
+      --add-opens=java.base/java.math=ALL-UNNAMED
+      --add-opens=java.base/java.net=ALL-UNNAMED
+      --add-opens=java.base/java.text=ALL-UNNAMED
+      --add-opens=java.base/java.util=ALL-UNNAMED
+      --add-opens=java.base/java.util.concurrent=ALL-UNNAMED
       --add-opens=java.base/java.util.zip=ALL-UNNAMED
       --add-opens=java.base/sun.security.util=ALL-UNNAMED
       --add-opens=java.base/sun.security.x509=ALL-UNNAMED
@@ -2743,6 +2754,16 @@
         </dependencies>
       </dependencyManagement>
     </profile>
+    <!-- We added this profile to support compilation for JDK 9 and above. -->
+    <profile>
+      <id>java9</id>
+      <activation>
+        <jdk>[9,)</jdk>
+      </activation>
+      <properties>
+        <maven.compiler.release>${javac.version}</maven.compiler.release>
+      </properties>
+    </profile>
   </profiles>
 
   <repositories>

hadoop-tools/hadoop-distcp/pom.xml

@@ -143,7 +143,7 @@
           <forkCount>1</forkCount>
           <reuseForks>false</reuseForks>
           <forkedProcessTimeoutInSeconds>600</forkedProcessTimeoutInSeconds>
-          <argLine>-Xmx1024m</argLine>
+          <argLine>${maven-surefire-plugin.argLine} -Xmx1024m</argLine>
           <includes>
             <include>**/Test*.java</include>
           </includes>

hadoop-tools/hadoop-federation-balance/pom.xml

@@ -147,7 +147,7 @@
           <forkCount>1</forkCount>
           <reuseForks>false</reuseForks>
           <forkedProcessTimeoutInSeconds>600</forkedProcessTimeoutInSeconds>
-          <argLine>-Xmx1024m</argLine>
+          <argLine>${maven-surefire-plugin.argLine} -Xmx1024m</argLine>
           <includes>
             <include>**/Test*.java</include>
           </includes>

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/ReservedContainerCandidatesSelector.java

@@ -170,6 +170,9 @@ private NodeForPreemption getPreemptionCandidatesOnNode(
       Map<ApplicationAttemptId, Set<RMContainer>> selectedCandidates,
       Resource totalPreemptionAllowed, boolean readOnly) {
     RMContainer reservedContainer = node.getReservedContainer();
+    if (reservedContainer == null) {
+      return null;
+    }
     Resource available = Resources.clone(node.getUnallocatedResource());
     Resource totalSelected = Resources.createResource(0);
     List<RMContainer> sortedRunningContainers =

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java

@@ -876,10 +876,9 @@ private void completeOustandingUpdatesWhichAreReserved(
       RMContainer rmContainer, ContainerStatus containerStatus,
       RMContainerEventType event) {
     N schedulerNode = getSchedulerNode(rmContainer.getNodeId());
-    if (schedulerNode != null &&
-        schedulerNode.getReservedContainer() != null) {
+    if (schedulerNode != null) {
       RMContainer resContainer = schedulerNode.getReservedContainer();
-      if (resContainer.getReservedSchedulerKey() != null) {
+      if (resContainer != null && resContainer.getReservedSchedulerKey() != null) {
         ContainerId containerToUpdate = resContainer
             .getReservedSchedulerKey().getContainerToUpdate();
         if (containerToUpdate != null &&

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/RegularContainerAllocator.java

@@ -858,12 +858,13 @@ private ContainerAllocation allocate(Resource clusterResource,
       FiCaSchedulerNode node = iter.next();
 
       // Do not schedule if there are any reservations to fulfill on the node
+      RMContainer nodeReservedContainer = node.getReservedContainer();
       if (iter.hasNext() &&
-          node.getReservedContainer() != null &&
+          nodeReservedContainer != null &&
           isSkipAllocateOnNodesWithReservedContainer()) {
         LOG.debug("Skipping scheduling on node {} since it has already been"
                 + " reserved by {}", node.getNodeID(),
-            node.getReservedContainer().getContainerId());
+            nodeReservedContainer.getContainerId());
         ActivitiesLogger.APP.recordSkippedAppActivityWithoutAllocation(
             activitiesManager, node, application, schedulerKey,
             ActivityDiagnosticConstant.NODE_HAS_BEEN_RESERVED, ActivityLevel.NODE);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java

@@ -520,13 +520,13 @@ public boolean accept(Resource cluster,
             // When reserve a resource (state == NEW is for new container,
             // state == RUNNING is for increase container).
             // Just check if the node is not already reserved by someone
-            if (schedulerContainer.getSchedulerNode().getReservedContainer()
-                != null) {
+            RMContainer reservedContainer =
+                schedulerContainer.getSchedulerNode().getReservedContainer();
+            if (reservedContainer != null) {
               if (LOG.isDebugEnabled()) {
                 LOG.debug("Try to reserve a container, but the node is "
                     + "already reserved by another container="
-                    + schedulerContainer.getSchedulerNode()
-                    .getReservedContainer().getContainerId());
+                    + reservedContainer.getContainerId());
               }
               return false;
             }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase-tests/pom.xml

@@ -491,6 +491,8 @@
       </plugin>
       <!-- The fork value is deliberately set to 0 to avoid VM crash while running tests
        on Jenkins, removing this leads to tests crashing silently due to VM crash -->
+      <!-- TODO: we should investigate and address the crash issue and re-enable fork,
+                 otherwise, JPMS args does not take effect -->
       <plugin>
         <artifactId>maven-surefire-plugin</artifactId>
         <configuration>