title: NullPointerException from Hadoop's JobSplitWriter / SerializationFactory when calling InputSplit's getClass()
body: <p>I'm getting a <code>NullPointerException</code> when launching a <code>MapReduce</code> job. It's being thrown by the <code>SerializationFactory</code>'s <code>getSerializer()</code> method. I'm using a custom <code>InputSplit</code>, <code>InputFormat</code>, <code>RecordReader</code> and <code>MapReduce</code> value class.</p>
  
  <p>I know the error is being thrown some time after the splits are being created by my <code>InputFormat</code> class, but before the creation of the <code>RecordReader</code>. As far as I can tell, it is occurring directly after the "cleaning up the staging area" message.</p>
  
  <p>By checking the Hadoop source in the places indicated by the stack trace, it looks like the error is occuring when <code>getSerialization()</code> receives a null <code>Class&lt;T&gt;</code> pointer. The JobClient's <code>writeNewSplits()</code> calls that method like this: </p>
  
  <pre><code>Serializer&lt;T&gt; serializer = factory.getSerializer((Class&lt;T&gt;) split.getClass());
  </code></pre>
  
  <p>Thus, I assume that when <code>getClass()</code> is being called on my custom <code>InputSplit</code> objects, it's returning a <code>null</code> pointer, but that is just baffling. Any ideas?</p>
  
  <p>The full stack trace from the error follows:</p>
  
  <pre>
  12/06/24 14:26:49 INFO mapred.JobClient: Cleaning up the staging area hdfs://localhost:54310/tmp/hadoop-s3cur3/mapred/staging/s3cur3/.staging/job_201206240915_0035
  Exception in thread "main" java.lang.NullPointerException
      at org.apache.hadoop.io.serializer.SerializationFactory.getSerializer(SerializationFactory.java:73)
      at org.apache.hadoop.mapreduce.split.JobSplitWriter.writeNewSplits(JobSplitWriter.java:123)
      at org.apache.hadoop.mapreduce.split.JobSplitWriter.createSplitFiles(JobSplitWriter.java:74)
      at org.apache.hadoop.mapred.JobClient.writeNewSplits(JobClient.java:968)
      at org.apache.hadoop.mapred.JobClient.writeSplits(JobClient.java:979)
      at org.apache.hadoop.mapred.JobClient.access$600(JobClient.java:174)
      at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:897)
      at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:850)
      at java.security.AccessController.doPrivileged(Native Method)
      at javax.security.auth.Subject.doAs(Subject.java:396)
      at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1121)
      at org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:850)
      at org.apache.hadoop.mapreduce.Job.submit(Job.java:500)
      at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:530)
      at edu.cs.illinois.cogcomp.hadoopinterface.infrastructure.CuratorJob.start(CuratorJob.java:94)
      at edu.cs.illinois.cogcomp.hadoopinterface.HadoopInterface.main(HadoopInterface.java:58)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
      at java.lang.reflect.Method.invoke(Method.java:597)
      at org.apache.hadoop.util.RunJar.main(RunJar.java:156)
  </pre>
  
  <p>Thanks!</p>
  
  <p>EDIT: My code for the custom InputSplit follows:</p>
  
  <pre><code>import . . .
  
  /**
   * A document directory within the input directory. 
   * Returned by DirectoryInputFormat.getSplits()
   * and passed to DirectoryInputFormat.createRecordReader().
   *
   * Represents the data to be processed by an individual Map process.
   */
  public class DirectorySplit extends InputSplit {
      /**
       * Constructs a DirectorySplit object
       * @param docDirectoryInHDFS The location (in HDFS) of this
       *            document's directory, complete with all annotations.
       * @param fs The filesystem associated with this job
       */
      public  DirectorySplit( Path docDirectoryInHDFS, FileSystem fs )
              throws IOException {
          this.inputPath = docDirectoryInHDFS;
          hash = FileSystemHandler.getFileNameFromPath(inputPath);
          this.fs = fs;
      }
  
      /**
       * Get the size of the split so that the input splits can be sorted by size.
       * Here, we calculate the size to be the number of bytes in the original
       * document (i.e., ignoring all annotations).
       *
       * @return The number of characters in the original document
       */
      @Override
      public long getLength() throws IOException, InterruptedException {
          Path origTxt = new Path( inputPath, "original.txt" );
          HadoopInterface.logger.log( msg );
          return FileSystemHandler.getFileSizeInBytes( origTxt, fs);
      }
  
      /**
       * Get the list of nodes where the data for this split would be local.
       * This list includes all nodes that contain any of the required data---it's
       * up to Hadoop to decide which one to use.
       *
       * @return An array of the nodes for whom the split is local
       * @throws IOException
       * @throws InterruptedException
       */
      @Override
      public String[] getLocations() throws IOException, InterruptedException {
          FileStatus status = fs.getFileStatus(inputPath);
  
          BlockLocation[] blockLocs = fs.getFileBlockLocations( status, 0,
                                                                status.getLen() );
  
          HashSet&lt;String&gt; allBlockHosts = new HashSet&lt;String&gt;();
          for( BlockLocation blockLoc : blockLocs ) {
              allBlockHosts.addAll( Arrays.asList( blockLoc.getHosts() ) );
          }
  
          return (String[])allBlockHosts.toArray();
      }
  
      /**
       * @return The hash of the document that this split handles
       */
      public String toString() {
          return hash;
      }
  
      private Path inputPath;
      private String hash;
      private FileSystem fs;
  }
  </code></pre>