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 testUnreachableKafkaBrokers() {
    // Expect an exception when the Kafka brokers are not reachable on the workers.
    // We specify partitions explicitly so that splitting does not involve server interaction.
    // Set request timeout to 10ms so that test does not take long.

    thrown.expect(Exception.class);
    thrown.expectMessage("Reader-0: Timeout while initializing partition 'test-0'");

    int numElements = 1000;
    PCollection<Long> input =
        p.apply(
                KafkaIO.<Integer, Long>read()
                    .withBootstrapServers("8.8.8.8:9092") // Google public DNS ip.
                    .withTopicPartitions(ImmutableList.of(new TopicPartition("test", 0)))
                    .withKeyDeserializer(IntegerDeserializer.class)
                    .withValueDeserializer(LongDeserializer.class)
                    .withConsumerConfigUpdates(
                        ImmutableMap.of(
                            ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,
                            5,
                            ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,
                            8,
                            ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
                            8,
                            "default.api.timeout.ms",
                            10))
                    .withMaxNumRecords(10)
                    .withoutMetadata())
            .apply(Values.create());

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