### Question:
BlurRecordWriter extends RecordWriter<Text, BlurRecord> { public BlurRecordWriter(TaskAttemptContext context) throws IOException { Configuration configuration = context.getConfiguration(); String outputPath = configuration.get("mapred.output.dir"); int id = context.getTaskAttemptID().getTaskID().getId(); String shardName = BlurUtil.getShardName(BlurConstants.SHARD_PREFIX, id); Path basePath = new Path(outputPath); Path indexPath = new Path(basePath, shardName); Analyzer analyzer = new KeywordAnalyzer(); IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_35, analyzer); Directory dir = new HdfsDirectory(indexPath); dir.setLockFactory(NoLockFactory.getNoLockFactory()); writer = new IndexWriter(dir, conf); } BlurRecordWriter(TaskAttemptContext context); @Override void write(Text key, BlurRecord value); @Override void close(TaskAttemptContext context);  }### Answer:
@Test public void testBlurRecordWriter() throws IOException, InterruptedException { JobID jobId = new JobID(); TaskID tId = new TaskID(jobId, false, 13); TaskAttemptID taskId = new TaskAttemptID(tId, 0); Configuration conf = new Configuration(); String pathStr = "./tmp/output-record-writer-test-newapi"; rm(new File(pathStr)); conf.set("mapred.output.dir", pathStr); TaskAttemptContext context = new TaskAttemptContext(conf, taskId); BlurRecordWriter writer = new BlurRecordWriter(context); Text key = new Text(); BlurRecord value = new BlurRecord(); for (int i = 0; i < 10; i++) { String rowId = UUID.randomUUID().toString(); key.set(rowId); value.setFamily("cf"); value.setRowId(rowId); value.setRecordId(UUID.randomUUID().toString()); value.addColumn("name", "value"); writer.write(key, value); } writer.close(context); HdfsDirectory dir = new HdfsDirectory(new Path(pathStr,BlurUtil.getShardName(BlurConstants.SHARD_PREFIX, 13))); assertTrue(IndexReader.indexExists(dir)); IndexReader reader = IndexReader.open(dir); assertEquals(10,reader.numDocs()); }