public static <K, V> Read<K, V> read() {
    return new AutoValue_KafkaIO_Read.Builder<K, V>()
        .setTopics(new ArrayList<>())
        .setTopicPartitions(new ArrayList<>())
        .setConsumerFactoryFn(KafkaIOUtils.KAFKA_CONSUMER_FACTORY_FN)
        .setConsumerConfig(KafkaIOUtils.DEFAULT_CONSUMER_PROPERTIES)
        .setMaxNumRecords(Long.MAX_VALUE)
        .setCommitOffsetsInFinalizeEnabled(false)
        .setDynamicRead(false)
        .setTimestampPolicyFactory(TimestampPolicyFactory.withProcessingTime())
        .setConsumerPollingTimeout(2L)
        .setRedistributed(false)
        .setAllowDuplicates(false)
        .setRedistributeNumKeys(0)
        .build();
  }
@Test
  public void testUnboundedReaderLogsCommitFailure() throws Exception {

    List<String> topics = ImmutableList.of("topic_a");

    PositionErrorConsumerFactory positionErrorConsumerFactory = new PositionErrorConsumerFactory();

    UnboundedSource<KafkaRecord<Integer, Long>, KafkaCheckpointMark> source =
        KafkaIO.<Integer, Long>read()
            .withBootstrapServers("myServer1:9092,myServer2:9092")
            .withTopics(topics)
            .withConsumerFactoryFn(positionErrorConsumerFactory)
            .withKeyDeserializer(IntegerDeserializer.class)
            .withValueDeserializer(LongDeserializer.class)
            .makeSource();

    UnboundedReader<KafkaRecord<Integer, Long>> reader = source.createReader(null, null);

    reader.start();

    unboundedReaderExpectedLogs.verifyWarn("exception while fetching latest offset for partition");

    reader.close();
  }