public static <T> Read<T> read() {
    return new AutoValue_KuduIO_Read.Builder<T>().setKuduService(new KuduServiceImpl<>()).build();
  }
@Test
  public void testRead() throws KuduException {
    when(mockReadService.createReader(any())).thenAnswer(new FakeReaderAnswer());
    // Simulate the equivalent of Kudu providing an encoded scanner per tablet. Here we encode
    // a range which the fake reader will use to simulate a single tablet read.
    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")
                // the fake reader only deals with a single int
                .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();

    // check that the fake tablet ranges were read
    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));
  }