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 testUnboundedSourceWithExceptionInKafkaFetch() {
    // Similar testUnboundedSource, but with an injected exception inside Kafk Consumer poll.

    // The reader should throw an IOException:
    thrown.expectCause(isA(IOException.class));
    thrown.expectCause(hasMessage(containsString("Exception while reading from Kafka")));
    // The original exception is from MockConsumer.poll():
    thrown.expectCause(hasCause(isA(KafkaException.class)));
    thrown.expectCause(hasCause(hasMessage(containsString("Injected error in consumer.poll()"))));

    int numElements = 1000;
    String topic = "my_topic";

    KafkaIO.Read<Integer, Long> reader =
        KafkaIO.<Integer, Long>read()
            .withBootstrapServers("none")
            .withTopic("my_topic")
            .withConsumerFactoryFn(
                new ConsumerFactoryFn(
                    ImmutableList.of(topic), 10, numElements, OffsetResetStrategy.EARLIEST))
            .withMaxNumRecords(2 * numElements) // Try to read more messages than available.
            .withConsumerConfigUpdates(ImmutableMap.of("inject.error.at.eof", true))
            .withKeyDeserializer(IntegerDeserializer.class)
            .withValueDeserializer(LongDeserializer.class);

    PCollection<Long> input = p.apply(reader.withoutMetadata()).apply(Values.create());

    addCountingAsserts(input, numElements);
    p.run();
  }