/*
 * Licensed to the University of California, Berkeley 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
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * 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 alluxio.client.keyvalue.hadoop;

import alluxio.AlluxioURI;
import alluxio.annotation.PublicApi;
import alluxio.client.ClientContext;
import alluxio.client.keyvalue.KeyValueMasterClient;
import alluxio.client.keyvalue.KeyValueSystem;
import alluxio.exception.AlluxioException;
import alluxio.thrift.PartitionInfo;

import com.google.common.collect.Lists;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;

import java.io.IOException;
import java.util.List;

import javax.annotation.concurrent.ThreadSafe;

/**
 * Implementation of {@link org.apache.hadoop.mapred.InputFormat} for MapReduce programs to access
 * {@link KeyValueSystem}.
 * <p>
 * It takes a {@link KeyValueSystem} URI, and emits key-value pairs stored in the KeyValueStore to
 * {@link org.apache.hadoop.mapred.Mapper}s.
 */
@PublicApi
@ThreadSafe
public final class KeyValueInputFormat implements InputFormat {
  private final KeyValueMasterClient mKeyValueMasterClient =
      new KeyValueMasterClient(ClientContext.getMasterAddress(), ClientContext.getConf());

  /**
   * Returns each partition as a {@link KeyValueInputSplit}.
   *
   * @param conf MapReduce job configuration
   * @param numSplits number of splits, ignored because it is determined by number of partitions
   * @return list of {@link InputSplit}s, each split is a partition
   * @throws IOException if information about the partition cannot be retrieved
   */
  @Override
  public InputSplit[] getSplits(JobConf conf, int numSplits) throws IOException {
    // The paths are MapReduce program's inputs specified in
    // {@code mapreduce.input.fileinputformat.inputdir}, each path should be a key-value store.
    Path[] paths = FileInputFormat.getInputPaths(conf);
    List<InputSplit> splits = Lists.newArrayList();
    try {
      for (Path path : paths) {
        List<PartitionInfo> partitionInfos =
            mKeyValueMasterClient.getPartitionInfo(new AlluxioURI(path.toString()));
        for (PartitionInfo partitionInfo : partitionInfos) {
          splits.add(new KeyValueInputSplit(partitionInfo));
        }
      }
    } catch (AlluxioException e) {
      throw new IOException(e);
    }
    InputSplit[] ret = new InputSplit[splits.size()];
    return splits.toArray(ret);
  }

  @Override
  public RecordReader getRecordReader(InputSplit inputSplit, JobConf jobConf, Reporter reporter)
      throws IOException {
    if (inputSplit instanceof KeyValueInputSplit) {
      try {
        return new KeyValueRecordReader((KeyValueInputSplit) inputSplit);
      } catch (AlluxioException e) {
        throw new IOException(e);
      }
    } else {
      throw new IOException("Expected InputSplit to be instance of KeyValueInputSplit.");
    }
  }
}

@@ -1212,29 +1212,70 @@
 oop.
-mapred.FileInputForma
+io.BytesWritable;%0Aimport org.apache.hadoop.mapreduce.InputSpli
 t;%0Ai
@@ -1304,28 +1304,32 @@
 p.mapred
-.InputFormat
+uce.RecordReader
 ;%0Aimport
@@ -1353,26 +1353,30 @@
 p.mapred
+uce
 .Input
-Spli
+Forma
 t;%0Aimpor
@@ -1405,16 +1405,22 @@
 pred
+uce
 .JobCon
-f
+text
 ;%0Aim
@@ -1448,29 +1448,38 @@
 p.mapred
-.RecordReader
+uce.TaskAttemptContext
 ;%0Aimport
@@ -1507,17 +1507,37 @@
 pred
-.Reporter
+uce.lib.input.FileInputFormat
 ;%0A%0Ai
@@ -1770,16 +1770,17 @@
 %7D.%0A * %3Cp
+/
 %3E%0A * It 
@@ -1987,17 +1987,14 @@
 mat 
-implement
+extend
 s In
@@ -2002,16 +2002,46 @@
 utFormat
+%3CBytesWritable, BytesWritable%3E
  %7B%0A  pri
@@ -2211,27 +2211,17 @@
 rns 
-each partition as a
+a list of
  %7B@l
@@ -2247,156 +2247,106 @@
 lit%7D
-.%0A   *%0A   * @param conf MapReduce job configuration%0A   * @param numSplits number of splits, ignored because it is determined by number of parti
+ where each split is one key-value partition.%0A   *%0A   * @param context MapReduce job configura
 tion
-s
 %0A   
@@ -2517,16 +2517,21 @@
  public 
+List%3C
 InputSpl
@@ -2532,18 +2532,17 @@
 putSplit
-%5B%5D
+%3E
  getSpli
@@ -2554,29 +2554,20 @@
 bCon
-f conf, int numSplits
+text context
 ) th
@@ -2794,17 +2794,20 @@
 aths(con
-f
+text
 );%0A    L
@@ -3246,321 +3246,97 @@
     
-InputSplit%5B%5D ret = new InputSplit%5Bsplits.size()%5D;%0A    return splits.toArray(ret);%0A  %7D%0A%0A  @Override%0A  public RecordReader getRecordReader(InputSplit inputSplit, JobConf jobConf, Reporter reporter)%0A      throws IOException %7B%0A    if (inputSplit instanceof KeyValueInputSplit) %7B%0A      try %7B%0A        return new KeyValu
+return splits;%0A  %7D%0A%0A  @Override%0A  public RecordReader%3CBytesWritable, BytesWritable%3E creat
 eRec
@@ -3349,17 +3349,8 @@
 der(
-(KeyValue
 Inpu
@@ -3355,17 +3355,16 @@
 putSplit
-)
  inputSp
@@ -3370,65 +3370,50 @@
 plit
-);
+,
 %0A      
-%7D catch (AlluxioException e) %7B%0A       
+TaskAttemptContext context)
  throw
- new
+s
  IOE
@@ -3424,129 +3424,49 @@
 tion
-(e);%0A      %7D%0A    %7D else %7B%0A      throw new IOException(%22Expected InputSplit to be instance of KeyValueInputSplit.%22);%0A    %7D
+ %7B%0A    return new KeyValueRecordReader();
 %0A  %7D

Update KeyValueInptuFormat to extend new hadoop API