[s] using key NullWritable with StructuredRecord value

format-xls/src/main/java/io/cdap/plugin/format/xls/input/XlsInputFormat.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.RecordReader;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
@@ -47,27 +48,25 @@
  * The {@link XlsInputFormat.XlsRecordReader} reads a given sheet, and within a sheet reads
  * all columns and all rows.
  */
-public class XlsInputFormat extends CombineFileInputFormat<LongWritable, StructuredRecord> {
+public class XlsInputFormat extends CombineFileInputFormat<NullWritable, StructuredRecord> {
 
   public static final String SHEET_NO = "Sheet Number";
   public static final String SHEET_VALUE = "sheetValue";
   public static final String NAME_SKIP_HEADER = "skipHeader";
   public static final String TERMINATE_IF_EMPTY_ROW = "terminateIfEmptyRow";
 
   @Override
-  public RecordReader<LongWritable, StructuredRecord> createRecordReader(InputSplit split, TaskAttemptContext context) {
+  public RecordReader<NullWritable, StructuredRecord> createRecordReader(InputSplit split, TaskAttemptContext context) {
     return new XlsRecordReader();
   }
 
   /**
    * Reads excel spread sheet, where the keys are the offset in the excel file and the text is the complete record.
    */
-  public static class XlsRecordReader extends RecordReader<LongWritable, StructuredRecord> {
+  public static class XlsRecordReader extends RecordReader<NullWritable, StructuredRecord> {
     // DataFormatter to format and get each cell's value as String
     XlsDataFormatter formatter;
     FormulaEvaluator formulaEvaluator;
-    // Map key that represents the row index.
-    private LongWritable key;
     // Map value that represents an excel row
     private StructuredRecord value;
     private Sheet workSheet;
@@ -141,7 +140,6 @@ public boolean nextKeyValue() {
           return true;
         }
       }
-      key = new LongWritable(rowIndex);
 
       StructuredRecord.Builder builder = StructuredRecord.builder(outputSchema);
       List<Schema.Field> fields = outputSchema.getFields();
@@ -188,8 +186,8 @@ public void close() throws IOException {
     }
 
     @Override
-    public LongWritable getCurrentKey() {
-      return key;
+    public NullWritable getCurrentKey() {
+      return NullWritable.get();
     }
 
     @Override