CoansysInputDocumentReader implements InputDocumentReader<String, DocumentWrapper>, Serializable { @Override public JavaPairRDD<String, DocumentWrapper> readDocuments(JavaSparkContext sparkContext, String inputDocumentPath) { Preconditions.checkArgument(StringUtils.isNotBlank(inputDocumentPath)); return sparkContext.sequenceFile(inputDocumentPath, Writable.class, BytesWritable.class) .mapToPair(bw -> bytesWritableConverter.convertToDocumentWrapperTuple2(bw._2())); }  @Override JavaPairRDD<String, DocumentWrapper> readDocuments(JavaSparkContext sparkContext, String inputDocumentPath); void setBytesWritableConverter(BytesWritableConverter bytesWritableConverter);  }
@Test(expectedExceptions=IllegalArgumentException.class) public void readDocuments_empty() { coansysInputDocumentReader.readDocuments(sparkContext, " "); }
@Test public void readDocuments() throws Exception { String inputDocumentPath = "/path"; when(sparkContext.sequenceFile(inputDocumentPath, Writable.class, BytesWritable.class)).thenReturn(rawDocuments); doReturn(docWrappers).when(rawDocuments).mapToPair(Mockito.any()); JavaPairRDD<String, DocumentWrapper> retDocWrappers = coansysInputDocumentReader.readDocuments(sparkContext, inputDocumentPath); assertTrue(docWrappers == retDocWrappers); verify(sparkContext).sequenceFile(inputDocumentPath, Writable.class, BytesWritable.class); verify(rawDocuments).mapToPair(convertToDocumentWrapperTuple2Function.capture()); assertConvertToDocumentWrapperTuple2Function(convertToDocumentWrapperTuple2Function.getValue()); }