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
  @Ignore // TODO : BEAM-4086 : enable once flakiness is fixed.
  public void testUnboundedSourceWithoutBoundedWrapper() {
    // This is same as testUnboundedSource() without the BoundedSource wrapper.
    // Most of the tests in this file set 'maxNumRecords' on the source, which wraps
    // the unbounded source in a bounded source. As a result, the test pipeline run as
    // bounded/batch pipelines under direct-runner.
    // This tests runs without such a wrapper over unbounded wrapper, and depends on watermark
    // progressing to infinity to end the test (see TimestampPolicyWithEndOfSource above).

    final int numElements = 1000;
    final int numPartitions = 10;
    String topic = "testUnboundedSourceWithoutBoundedWrapper";

    KafkaIO.Read<byte[], Long> reader =
        KafkaIO.<byte[], Long>read()
            .withBootstrapServers(topic)
            .withTopic(topic)
            .withConsumerFactoryFn(
                new ConsumerFactoryFn(
                    ImmutableList.of(topic),
                    numPartitions,
                    numElements,
                    OffsetResetStrategy.EARLIEST))
            .withKeyDeserializer(ByteArrayDeserializer.class)
            .withValueDeserializer(LongDeserializer.class)
            .withTimestampPolicyFactory(
                new TimestampPolicyWithEndOfSource<>(numElements / numPartitions - 1));

    p.apply("readFromKafka", reader.withoutMetadata())
        .apply(Values.create())
        .apply(Window.into(FixedWindows.of(Duration.standardDays(100))));

    PipelineResult result = p.run();

    MetricName elementsRead = SourceMetrics.elementsRead().getName();

    MetricQueryResults metrics =
        result
            .metrics()
            .queryMetrics(
                MetricsFilter.builder()
                    .addNameFilter(MetricNameFilter.inNamespace(elementsRead.getNamespace()))
                    .build());

    assertThat(
        metrics.getCounters(),
        hasItem(
            attemptedMetricsResult(
                elementsRead.getNamespace(),
                elementsRead.getName(),
                "readFromKafka",
                (long) numElements)));
  }