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 testResolveDefaultApiTimeout() {

    final String defaultApiTimeoutConfig = "default.api.timeout.ms";

    assertEquals(
        Duration.millis(20),
        KafkaUnboundedReader.resolveDefaultApiTimeout(
            KafkaIO.<Integer, Long>read()
                .withConsumerConfigUpdates(
                    ImmutableMap.of(
                        defaultApiTimeoutConfig,
                        20,
                        ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,
                        30))));

    assertEquals(
        Duration.millis(2 * 30),
        KafkaUnboundedReader.resolveDefaultApiTimeout(
            KafkaIO.<Integer, Long>read()
                .withConsumerConfigUpdates(
                    ImmutableMap.of(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30))));

    assertEquals(
        Duration.millis(60 * 1000),
        KafkaUnboundedReader.resolveDefaultApiTimeout(KafkaIO.<Integer, Long>read()));
  }