KuduIO { public static <T> Read<T> read() { return new AutoValue_KuduIO_Read.Builder<T>().setKuduService(new KuduServiceImpl<>()).build(); } private  KuduIO(); static Read<T> read(); static Write<T> write();  }
@Test public void testRead() throws KuduException { when(mockReadService.createReader(any())).thenAnswer(new FakeReaderAnswer()); List<byte[]> fakeScanners = Arrays.asList( ByteBuffer.allocate(8).putInt(0).putInt(25).array(), ByteBuffer.allocate(8).putInt(25).putInt(50).array(), ByteBuffer.allocate(8).putInt(50).putInt(75).array(), ByteBuffer.allocate(8).putInt(75).putInt(100).array()); when(mockReadService.createTabletScanners(any())).thenReturn(fakeScanners); PCollection<Integer> output = readPipeline.apply( KuduIO.<Integer>read() .withMasterAddresses("mock") .withTable("Table") .withParseFn( (SerializableFunction<RowResult, Integer>) input -> input.getInt(COL_ID)) .withKuduService(mockReadService) .withCoder(BigEndianIntegerCoder.of())); PAssert.thatSingleton(output.apply("Count", Count.globally())).isEqualTo((long) 100); readPipeline.run().waitUntilFinish(); expectedReadLogs.verifyDebug(String.format(FakeReader.LOG_SET_RANGE, 0, 25)); expectedReadLogs.verifyDebug(String.format(FakeReader.LOG_SET_RANGE, 25, 50)); expectedReadLogs.verifyDebug(String.format(FakeReader.LOG_SET_RANGE, 50, 75)); expectedReadLogs.verifyDebug(String.format(FakeReader.LOG_SET_RANGE, 75, 100)); }