Merge branch 'trunk' of github.com:apache/hadoop into HADOOP-15984-trunk-2

Author: virajjasani

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/FadvisedFileRegion.java

@@ -41,6 +41,7 @@ public class FadvisedFileRegion extends DefaultFileRegion {
   private static final Logger LOG =
       LoggerFactory.getLogger(FadvisedFileRegion.class);
 
+  private final Object closeLock = new Object();
   private final boolean manageOsCache;
   private final int readaheadLength;
   private final ReadaheadPool readaheadPool;
@@ -51,12 +52,12 @@ public class FadvisedFileRegion extends DefaultFileRegion {
   private final int shuffleBufferSize;
   private final boolean shuffleTransferToAllowed;
   private final FileChannel fileChannel;
-  
-  private ReadaheadRequest readaheadRequest;
+
+  private volatile ReadaheadRequest readaheadRequest;
 
   public FadvisedFileRegion(RandomAccessFile file, long position, long count,
       boolean manageOsCache, int readaheadLength, ReadaheadPool readaheadPool,
-      String identifier, int shuffleBufferSize, 
+      String identifier, int shuffleBufferSize,
       boolean shuffleTransferToAllowed) throws IOException {
     super(file.getChannel(), position, count);
     this.manageOsCache = manageOsCache;
@@ -73,97 +74,110 @@ public FadvisedFileRegion(RandomAccessFile file, long position, long count,
 
   @Override
   public long transferTo(WritableByteChannel target, long position)
-      throws IOException {
-    if (readaheadPool != null && readaheadLength > 0) {
-      readaheadRequest = readaheadPool.readaheadStream(identifier, fd,
-          position() + position, readaheadLength,
-          position() + count(), readaheadRequest);
+          throws IOException {
+    synchronized (closeLock) {
+      if (fd.valid()) {
+        if (readaheadPool != null && readaheadLength > 0) {
+          readaheadRequest = readaheadPool.readaheadStream(identifier, fd,
+                  position() + position, readaheadLength,
+                  position() + count(), readaheadRequest);
+        }
+
+        if(this.shuffleTransferToAllowed) {
+          return super.transferTo(target, position);
+        } else {
+          return customShuffleTransfer(target, position);
+        }
+      } else {
+        return 0L;
+      }
     }
-    
-    if(this.shuffleTransferToAllowed) {
-      return super.transferTo(target, position);
-    } else {
-      return customShuffleTransfer(target, position);
-    } 
+
   }
 
   /**
-   * This method transfers data using local buffer. It transfers data from 
-   * a disk to a local buffer in memory, and then it transfers data from the 
+   * This method transfers data using local buffer. It transfers data from
+   * a disk to a local buffer in memory, and then it transfers data from the
    * buffer to the target. This is used only if transferTo is disallowed in
-   * the configuration file. super.TransferTo does not perform well on Windows 
-   * due to a small IO request generated. customShuffleTransfer can control 
-   * the size of the IO requests by changing the size of the intermediate 
+   * the configuration file. super.TransferTo does not perform well on Windows
+   * due to a small IO request generated. customShuffleTransfer can control
+   * the size of the IO requests by changing the size of the intermediate
    * buffer.
    */
   @VisibleForTesting
   long customShuffleTransfer(WritableByteChannel target, long position)
-      throws IOException {
+          throws IOException {
     long actualCount = this.count - position;
     if (actualCount < 0 || position < 0) {
       throw new IllegalArgumentException(
-          "position out of range: " + position +
-          " (expected: 0 - " + (this.count - 1) + ')');
+              "position out of range: " + position +
+                      " (expected: 0 - " + (this.count - 1) + ')');
     }
     if (actualCount == 0) {
       return 0L;
     }
-    
+
     long trans = actualCount;
     int readSize;
     ByteBuffer byteBuffer = ByteBuffer.allocate(
-        Math.min(
-            this.shuffleBufferSize,
-            trans > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) trans));
-    
+            Math.min(
+                    this.shuffleBufferSize,
+                    trans > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) trans));
+
     while(trans > 0L &&
-        (readSize = fileChannel.read(byteBuffer, this.position+position)) > 0) {
+            (readSize = fileChannel.read(byteBuffer, this.position+position)) > 0) {
       //adjust counters and buffer limit
       if(readSize < trans) {
         trans -= readSize;
         position += readSize;
         byteBuffer.flip();
       } else {
-        //We can read more than we need if the actualCount is not multiple 
+        //We can read more than we need if the actualCount is not multiple
         //of the byteBuffer size and file is big enough. In that case we cannot
         //use flip method but we need to set buffer limit manually to trans.
         byteBuffer.limit((int)trans);
         byteBuffer.position(0);
-        position += trans; 
+        position += trans;
         trans = 0;
       }
-      
+
       //write data to the target
       while(byteBuffer.hasRemaining()) {
         target.write(byteBuffer);
       }
-      
+
       byteBuffer.clear();
     }
-    
+
     return actualCount - trans;
   }
 
-  
+
   @Override
   protected void deallocate() {
-    if (readaheadRequest != null) {
-      readaheadRequest.cancel();
+    synchronized (closeLock) {
+      if (readaheadRequest != null) {
+        readaheadRequest.cancel();
+        readaheadRequest = null;
+      }
+      super.deallocate();
     }
-    super.deallocate();
   }
-  
+
   /**
    * Call when the transfer completes successfully so we can advise the OS that
    * we don't need the region to be cached anymore.
    */
   public void transferSuccessful() {
-    if (manageOsCache && count() > 0) {
-      try {
-        NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(identifier,
-            fd, position(), count(), POSIX_FADV_DONTNEED);
-      } catch (Throwable t) {
-        LOG.warn("Failed to manage OS cache for " + identifier, t);
+    synchronized (closeLock) {
+      if (fd.valid() && manageOsCache && count() > 0) {
+        try {
+          NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(identifier,
+                  fd, position(), count(), POSIX_FADV_DONTNEED);
+        } catch (Throwable t) {
+          LOG.warn("Failed to manage OS cache for " + identifier +
+                  " fd " + fd, t);
+        }
       }
     }
   }

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

@@ -1056,7 +1056,7 @@ private void copyPlacementQueueToSubmissionContext(
       LOG.info("Placed application with ID " + context.getApplicationId() +
           " in queue: " + placementContext.getQueue() +
           ", original submission queue was: " + context.getQueue());
-      context.setQueue(placementContext.getQueue());
+      context.setQueue(placementContext.getFullQueuePath());
     }
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.java

@@ -283,7 +283,7 @@ public void setUp() throws IOException {
     setupDispatcher(rmContext, conf);
   }
 
-  private static PlacementManager createMockPlacementManager(
+  public static PlacementManager createMockPlacementManager(
       String userRegex, String placementQueue, String placementParentQueue
   ) throws YarnException {
     PlacementManager placementMgr = mock(PlacementManager.class);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestWorkPreservingRMRestart.java

@@ -1651,6 +1651,97 @@ public void testUAMRecoveryOnRMWorkPreservingRestart() throws Exception {
     assertUnmanagedAMQueueMetrics(qm2, 1, 0, 0, 1);
   }
 
+  //   Test behavior of an app if two same name leaf queue with different queuePath
+  //   during work preserving rm restart with %specified mapping Placement Rule.
+  //   Test case does following:
+  //1. Submit an apps to queue root.joe.test.
+  //2. While the applications is running, restart the rm and
+  //   check whether the app submitted to the queue it was submitted initially.
+  //3. Verify that application running successfully.
+  @Test(timeout = 60000)
+  public void testQueueRecoveryOnRMWorkPreservingRestart() throws Exception {
+    if (getSchedulerType() != SchedulerType.CAPACITY) {
+      return;
+    }
+    CapacitySchedulerConfiguration csConf = new CapacitySchedulerConfiguration(conf);
+
+    csConf.setQueues(
+            CapacitySchedulerConfiguration.ROOT, new String[] {"default", "joe", "john"});
+    csConf.setCapacity(
+            CapacitySchedulerConfiguration.ROOT + "." + "joe", 25);
+    csConf.setCapacity(
+            CapacitySchedulerConfiguration.ROOT + "." + "john", 25);
+    csConf.setCapacity(
+            CapacitySchedulerConfiguration.ROOT + "." + "default", 50);
+
+    final String q1 = CapacitySchedulerConfiguration.ROOT + "." + "joe";
+    final String q2 = CapacitySchedulerConfiguration.ROOT + "." + "john";
+    csConf.setQueues(q1, new String[] {"test"});
+    csConf.setQueues(q2, new String[] {"test"});
+    csConf.setCapacity(
+            CapacitySchedulerConfiguration.ROOT + "." + "joe.test", 100);
+    csConf.setCapacity(
+            CapacitySchedulerConfiguration.ROOT + "." + "john.test", 100);
+
+    csConf.set(CapacitySchedulerConfiguration.MAPPING_RULE_JSON,
+        "{\"rules\" : [{\"type\": \"user\", \"policy\" : \"specified\", " +
+        "\"fallbackResult\" : \"skip\", \"matches\" : \"*\"}]}");
+
+    // start RM
+    rm1 = new MockRM(csConf);
+    rm1.start();
+    MockMemoryRMStateStore memStore =
+        (MockMemoryRMStateStore) rm1.getRMStateStore();
+    MockNM nm1 =
+        new MockNM("127.0.0.1:1234", 15120, rm1.getResourceTrackerService());
+    nm1.registerNode();
+
+    RMContext newMockRMContext = rm1.getRMContext();
+    newMockRMContext.setQueuePlacementManager(TestAppManager.createMockPlacementManager(
+        "user1|user2", "test", "root.joe"));
+
+    MockRMAppSubmissionData data =
+        MockRMAppSubmissionData.Builder.createWithMemory(1024, rm1)
+            .withAppName("app")
+            .withQueue("root.joe.test")
+            .withUser("user1")
+            .withAcls(null)
+            .build();
+
+    RMApp app = MockRMAppSubmitter.submit(rm1, data);
+    MockAM am = MockRM.launchAndRegisterAM(app, rm1, nm1);
+    rm1.waitForState(app.getApplicationId(), RMAppState.RUNNING);
+
+    MockRM rm2 = new MockRM(csConf, memStore) {
+      @Override
+      protected RMAppManager createRMAppManager() {
+        return new RMAppManager(this.rmContext, this.scheduler,
+            this.masterService, this.applicationACLsManager, conf) {
+          @Override
+          ApplicationPlacementContext placeApplication(
+              PlacementManager placementManager,
+              ApplicationSubmissionContext context, String user,
+              boolean isRecovery) throws YarnException {
+            return super.placeApplication(
+                    newMockRMContext.getQueuePlacementManager(), context, user, isRecovery);
+          }
+        };
+      }
+    };
+
+    nm1.setResourceTrackerService(rm2.getResourceTrackerService());
+    rm2.start();
+    RMApp recoveredApp0 =
+        rm2.getRMContext().getRMApps().get(app.getApplicationId());
+
+    rm2.waitForState(recoveredApp0.getApplicationId(), RMAppState.ACCEPTED);
+    am.setAMRMProtocol(rm2.getApplicationMasterService(), rm2.getRMContext());
+    am.registerAppAttempt(true);
+    rm2.waitForState(recoveredApp0.getApplicationId(), RMAppState.RUNNING);
+
+    Assert.assertEquals("root.joe.test", recoveredApp0.getQueue());
+  }
+
   private void assertUnmanagedAMQueueMetrics(QueueMetrics qm, int appsSubmitted,
       int appsPending, int appsRunning, int appsCompleted) {
     Assert.assertEquals(appsSubmitted, qm.getUnmanagedAppsSubmitted());