public StreamXmlRecordReader(FSDataInputStream in, FileSplit split,
      TaskAttemptContext context, Configuration conf, FileSystem fs)
  throws IOException {
    super(in, split, context, conf, fs);

    beginMark_ = checkJobGet(CONF_NS + "begin");
    endMark_ = checkJobGet(CONF_NS + "end");

    maxRecSize_ = conf_.getInt(CONF_NS + "maxrec", 50 * 1000);
    lookAhead_ = conf_.getInt(CONF_NS + "lookahead", 2 * maxRecSize_);
    synched_ = false;

    slowMatch_ = conf_.getBoolean(CONF_NS + "slowmatch", false);
    if (slowMatch_) {
      beginPat_ = makePatternCDataOrMark(beginMark_);
      endPat_ = makePatternCDataOrMark(endMark_);
    }
    init();
  }
@Test
  public void testStreamXmlRecordReader() throws Exception {

    Job job = Job.getInstance();
    Configuration conf = job.getConfiguration();
    job.setJarByClass(TestStreamXmlRecordReader.class);
    job.setMapperClass(Mapper.class);
    conf.set("stream.recordreader.class",
        "org.apache.hadoop.streaming.mapreduce.StreamXmlRecordReader");
    conf.set("stream.recordreader.begin", "<PATTERN>");
    conf.set("stream.recordreader.end", "</PATTERN>");
    job.setInputFormatClass(StreamInputFormat.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileInputFormat.addInputPath(job, new Path("target/input.xml"));
    OUTPUT_DIR = new Path("target/output");
    fs = FileSystem.get(conf);
    if (fs.exists(OUTPUT_DIR)) {
      fs.delete(OUTPUT_DIR, true);
    }
    FileOutputFormat.setOutputPath(job, OUTPUT_DIR);
    boolean ret = job.waitForCompletion(true);

    assertEquals(true, ret);
    checkOutput();

  }