Fix findbugs warnings in mr-client modules part 3 (mahadev)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/MR-279@1155099 13f79535-47bb-0310-9956-ffa450edef68

mapreduce/CHANGES.txt

@@ -4,6 +4,8 @@ Trunk (unreleased changes)
 
   MAPREDUCE-279
  
+    Fix findbugs warnings in mr-client modules part 3 (mahadev)
+
     Fix findbugs warnings in mr-client modules part 2 (mahadev)
  
     Fix findbugs warnings in mr-client modules, part 1  (mahadev) 

mapreduce/dev-support/findbugs-exclude.xml

@@ -440,5 +440,20 @@
   <Match>
     <Class name="~org\.apache\.hadoop\.mapreduce\.v2\.proto.*" />
   </Match>
-
+  
+   <Match>
+     <Class name="org.apache.hadoop.mapreduce.v2.app.rm.RMContainerAllocator" />
+      <Field name="mapResourceReqt" />
+     <Bug pattern="IS2_INCONSISTENT_SYNC" />
+   </Match>
+  
+   <Match>
+     <Class name="org.apache.hadoop.mapreduce.v2.app.rm.RMContainerAllocator" />
+      <Field name="reduceResourceReqt" />
+     <Bug pattern="IS2_INCONSISTENT_SYNC" />
+   </Match>
+   <!--
+     The above 2 fields are accessed locally and only via methods that are synchronized. 
+     -->
+  
  </FindBugsFilter>

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/LocalContainerLauncher.java

@@ -439,7 +439,7 @@ FIXME:  do we really need to worry about renamed map outputs, or already moved t
               // this is recursive, unlike File delete():
               deleted = curFC.delete(new Path(curLocalFiles[j].getName()),true);
             }
-          } catch (Exception e) {
+          } catch (IOException e) {
             deleted = false;
           }
           if (!deleted) {

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/client/MRClientService.java

@@ -31,7 +31,6 @@ import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.mapreduce.JobACL;
 import org.apache.hadoop.mapreduce.v2.api.MRClientProtocol;
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.FailTaskAttemptRequest;
@@ -187,10 +186,11 @@ public class MRClientService extends AbstractService
       if (job == null) {
         throw RPCUtil.getRemoteException("Unknown job " + jobID);
       }
-      JobACL operation = JobACL.VIEW_JOB;
-      if (modifyAccess) {
-        operation = JobACL.MODIFY_JOB;
-      }
+      //TODO fix job acls.
+      //JobACL operation = JobACL.VIEW_JOB;
+      //if (modifyAccess) {
+      //  operation = JobACL.MODIFY_JOB;
+      //}
       //TO disable check access ofr now.
       //checkAccess(job, operation);
       return job;

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java

@@ -842,9 +842,13 @@ public class JobImpl implements org.apache.hadoop.mapreduce.v2.app.job.Job,
           attemptID.getTaskId().setTaskType(TaskType.MAP);
           TaskAttemptContext taskContext = new TaskAttemptContextImpl(job.conf,
               TypeConverter.fromYarn(attemptID));
-          OutputFormat outputFormat = ReflectionUtils.newInstance(
-              taskContext.getOutputFormatClass(), job.conf);
-          job.committer = outputFormat.getOutputCommitter(taskContext);
+          try {
+            OutputFormat outputFormat = ReflectionUtils.newInstance(
+                taskContext.getOutputFormatClass(), job.conf);
+            job.committer = outputFormat.getOutputCommitter(taskContext);
+          } catch(Exception e) {
+            throw new IOException("Failed to assign outputcommitter", e);
+          }
         } else {
           job.jobContext = new org.apache.hadoop.mapred.JobContextImpl(
               new JobConf(job.conf), job.oldJobId);
@@ -950,7 +954,7 @@ public class JobImpl implements org.apache.hadoop.mapreduce.v2.app.job.Job,
         return JobState.INITED;
         //TODO XXX Should JobInitedEvent be generated here (instead of in StartTransition)
 
-      } catch (Exception e) {
+      } catch (IOException e) {
         LOG.warn("Job init failed", e);
         job.addDiagnostic("Job init failed : "
             + StringUtils.stringifyException(e));

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java

@@ -155,7 +155,7 @@ public abstract class TaskAttemptImpl implements
   private Token<JobTokenIdentifier> jobToken;
   private static AtomicBoolean initialClasspathFlag = new AtomicBoolean();
   private static String initialClasspath = null;
-
+  private final Object classpathLock = new Object();
   private long launchTime;
   private long finishTime;
   private WrappedProgressSplitsBlock progressSplitBlock;
@@ -508,7 +508,7 @@ public abstract class TaskAttemptImpl implements
    * a parent CLC and use it for all the containers.
    */
   private String getInitialClasspath() throws IOException {
-    synchronized (initialClasspathFlag) {
+    synchronized (classpathLock) {
       if (initialClasspathFlag.get()) {
         return initialClasspath;
       }

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/recover/RecoveryService.java

@@ -111,9 +111,7 @@ public class RecoveryService extends CompositeService implements Recovery {
     this.startCount = startCount;
     this.dispatcher = new RecoveryDispatcher();
     this.clock = new ControlledClock(clock);
-    if (dispatcher instanceof Service) {
       addService((Service) dispatcher);
-    }
   }
 
   @Override

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMCommunicator.java

@@ -272,6 +272,9 @@ public class RMCommunicator extends AbstractService  {
     allocateRequest.addAllReleases(new ArrayList<ContainerId>());
     AllocateResponse allocateResponse = scheduler.allocate(allocateRequest);
     AMResponse response = allocateResponse.getAMResponse();
+    if (response.getReboot()) {
+      LOG.info("Event from RM: shutting down Application Master");
+    }
   }
 
 }

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java

@@ -30,8 +30,8 @@ import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.Map.Entry;
+import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -54,7 +54,6 @@ import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
 import org.apache.hadoop.yarn.api.records.AMResponse;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.api.records.ContainerState;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.util.RackResolver;
@@ -394,7 +393,10 @@ public class RMContainerAllocator extends RMContainerRequestor
     }
   }
 
-  private String getStat() {
+  /**
+   * Synchronized to avoid findbugs warnings
+   */
+  private synchronized String getStat() {
     return "PendingReduces:" + pendingReduces.size() +
         " ScheduledMaps:" + scheduledRequests.maps.size() +
         " ScheduledReduces:" + scheduledRequests.reduces.size() +

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/speculate/DefaultSpeculator.java

@@ -267,7 +267,7 @@ public class DefaultSpeculator extends AbstractService implements
       case TASK_CONTAINER_NEED_UPDATE:
       {
         AtomicInteger need = containerNeed(event.getTaskID());
-        int newNeed = need.addAndGet(event.containersNeededChange());
+        need.addAndGet(event.containersNeededChange());
         break;
       }
 

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/speculate/LegacyTaskRuntimeEstimator.java

@@ -22,8 +22,6 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.mapreduce.v2.api.records.JobId;
 import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId;
 import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptState;
@@ -40,7 +38,7 @@ public class LegacyTaskRuntimeEstimator extends StartEndTimesBase {
 
   private final Map<TaskAttempt, AtomicLong> attemptRuntimeEstimates
       = new ConcurrentHashMap<TaskAttempt, AtomicLong>();
-  private final Map<TaskAttempt, AtomicLong> attemptRuntimeEstimateVariances
+  private final ConcurrentHashMap<TaskAttempt, AtomicLong> attemptRuntimeEstimateVariances
       = new ConcurrentHashMap<TaskAttempt, AtomicLong>();
 
   @Override
@@ -93,16 +91,11 @@ public class LegacyTaskRuntimeEstimator extends StartEndTimesBase {
       }
 
       if (estimateVarianceContainer == null) {
-        synchronized (attemptRuntimeEstimateVariances) {
-          if (attemptRuntimeEstimateVariances.get(taskAttempt) == null) {
-            attemptRuntimeEstimateVariances.put(taskAttempt, new AtomicLong());
-          }
-
-          estimateVarianceContainer
-              = attemptRuntimeEstimateVariances.get(taskAttempt);
-        }
+        attemptRuntimeEstimateVariances.putIfAbsent(taskAttempt, new AtomicLong());
+        estimateVarianceContainer = attemptRuntimeEstimateVariances.get(taskAttempt);
       }
 
+
       long estimate = -1;
       long varianceEstimate = -1;
 
@@ -115,7 +108,9 @@ public class LegacyTaskRuntimeEstimator extends StartEndTimesBase {
       if (estimateContainer != null) {
         estimateContainer.set(estimate);
       }
-      estimateVarianceContainer.set(varianceEstimate);
+      if (estimateVarianceContainer != null) {
+        estimateVarianceContainer.set(varianceEstimate);
+      }
     }
   }
 

mapreduce/mr-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/jobhistory/JobHistoryUtils.java

@@ -140,7 +140,7 @@ public class JobHistoryUtils {
    * @throws IOException if the filename format is invalid.
    */
   public static JobID getJobIDFromHistoryFilePath(String pathString) throws IOException {
-    String [] parts = pathString.split(File.separator);
+    String [] parts = pathString.split(Path.SEPARATOR);
     String fileNamePart = parts[parts.length -1];
     JobIndexInfo jobIndexInfo =  FileNameIndexUtils.getIndexInfo(fileNamePart);
     return TypeConverter.fromYarn(jobIndexInfo.getJobId());
@@ -293,8 +293,8 @@ public class JobHistoryUtils {
     Matcher matcher = TIMESTAMP_DIR_PATTERN.matcher(path);
     if (matcher.find()) {
       String matched = matcher.group();
-      matched.intern();
-      return matched;
+      String ret = matched.intern();
+      return ret;
     } else {
       return null;
     }

mapreduce/mr-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRApps.java

@@ -156,41 +156,49 @@ public class MRApps extends Apps {
 
   public static void setInitialClasspath(
       Map<String, String> environment) throws IOException {
-
-    // Get yarn mapreduce-app classpath from generated classpath
-    // Works if compile time env is same as runtime. Mainly tests.
-    ClassLoader thisClassLoader =
-        Thread.currentThread().getContextClassLoader();
-    String mrAppGeneratedClasspathFile = "mrapp-generated-classpath";
-    InputStream classpathFileStream =
-        thisClassLoader.getResourceAsStream(mrAppGeneratedClasspathFile);
-    BufferedReader reader =
-        new BufferedReader(new InputStreamReader(classpathFileStream));
-    addToClassPath(environment, reader.readLine().trim());
-    // Put the file itself on classpath for tasks.
-    addToClassPath(environment,
-        thisClassLoader.getResource(mrAppGeneratedClasspathFile).getFile());
-
-    // If runtime env is different.
-    if (System.getenv().get("YARN_HOME") != null) {
-      ShellCommandExecutor exec =
-         new ShellCommandExecutor(new String[] {
-              System.getenv().get("YARN_HOME") + "/bin/yarn",
-              "classpath" });
-      exec.execute();
-      addToClassPath(environment, exec.getOutput().trim());
-    }
-
-    // Get yarn mapreduce-app classpath
-    if (System.getenv().get("HADOOP_MAPRED_HOME")!= null) {
-      ShellCommandExecutor exec =
-          new ShellCommandExecutor(new String[] {
-              System.getenv().get("HADOOP_MAPRED_HOME") + "/bin/mapred",
-              "classpath" });
-      exec.execute();
-      addToClassPath(environment, exec.getOutput().trim());
+    InputStream classpathFileStream = null;
+    try {
+      // Get yarn mapreduce-app classpath from generated classpath
+      // Works if compile time env is same as runtime. Mainly tests.
+      ClassLoader thisClassLoader =
+          Thread.currentThread().getContextClassLoader();
+      String mrAppGeneratedClasspathFile = "mrapp-generated-classpath";
+      classpathFileStream =
+          thisClassLoader.getResourceAsStream(mrAppGeneratedClasspathFile);
+      BufferedReader reader =
+          new BufferedReader(new InputStreamReader(classpathFileStream));
+      String cp = reader.readLine();
+      if (cp != null) {
+        addToClassPath(environment, cp.trim());
+      }
+      // Put the file itself on classpath for tasks.
+      addToClassPath(environment,
+          thisClassLoader.getResource(mrAppGeneratedClasspathFile).getFile());
+
+      // If runtime env is different.
+      if (System.getenv().get("YARN_HOME") != null) {
+        ShellCommandExecutor exec =
+            new ShellCommandExecutor(new String[] {
+                System.getenv().get("YARN_HOME") + "/bin/yarn",
+            "classpath" });
+        exec.execute();
+        addToClassPath(environment, exec.getOutput().trim());
+      }
+
+      // Get yarn mapreduce-app classpath
+      if (System.getenv().get("HADOOP_MAPRED_HOME")!= null) {
+        ShellCommandExecutor exec =
+            new ShellCommandExecutor(new String[] {
+                System.getenv().get("HADOOP_MAPRED_HOME") + "/bin/mapred",
+            "classpath" });
+        exec.execute();
+        addToClassPath(environment, exec.getOutput().trim());
+      }
+    } finally {
+      if (classpathFileStream != null) {
+        classpathFileStream.close();
+      }
     }
-
     // TODO: Remove duplicates.
   }