KuduIO { public static <T> Write<T> write() { return new AutoValue_KuduIO_Write.Builder<T>().setKuduService(new KuduServiceImpl<>()).build(); } private  KuduIO(); static Read<T> read(); static Write<T> write();  }
@Test @Ignore public void testWrite() throws Exception { when(mockWriteService.createWriter(any())).thenReturn(new FakeWriter()); writePipeline .apply("Generate sequence", GenerateSequence.from(0).to(numberRecords)) .apply( "Write records to Kudu", KuduIO.write() .withMasterAddresses("ignored") .withTable("ignored") .withFormatFn(new GenerateUpsert()) .withKuduService(mockWriteService)); writePipeline.run().waitUntilFinish(); for (int i = 1; i <= targetParallelism; i++) { expectedWriteLogs.verifyDebug(String.format(FakeWriter.LOG_OPEN_SESSION, i)); expectedWriteLogs.verifyDebug( String.format(FakeWriter.LOG_WRITE, i)); expectedWriteLogs.verifyDebug(String.format(FakeWriter.LOG_CLOSE_SESSION, i)); } for (int n = 0; n < numberRecords; n++) { expectedWriteLogs.verifyDebug( String.format(FakeWriter.LOG_WRITE_VALUE, n)); } }