public static <V> Read<V> read() {
    return new AutoValue_SparkReceiverIO_Read.Builder<V>().build();
  }
@Test
  public void testReadFromCustomReceiverWithOffset() {
    CustomReceiverWithOffset.shouldFailInTheMiddle = false;
    ReceiverBuilder<String, CustomReceiverWithOffset> receiverBuilder =
        new ReceiverBuilder<>(CustomReceiverWithOffset.class).withConstructorArgs();
    SparkReceiverIO.Read<String> reader =
        SparkReceiverIO.<String>read()
            .withGetOffsetFn(Long::valueOf)
            .withTimestampFn(Instant::parse)
            .withPullFrequencySec(PULL_FREQUENCY_SEC)
            .withStartPollTimeoutSec(START_POLL_TIMEOUT_SEC)
            .withStartOffset(START_OFFSET)
            .withSparkReceiverBuilder(receiverBuilder);

    List<String> expected = new ArrayList<>();
    for (int i = 0; i < CustomReceiverWithOffset.RECORDS_COUNT; i++) {
      expected.add(String.valueOf(i));
    }
    PCollection<String> actual = pipeline.apply(reader).setCoder(StringUtf8Coder.of());

    PAssert.that(actual).containsInAnyOrder(expected);
    pipeline.run().waitUntilFinish(Duration.standardSeconds(15));
  }