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 testUnboundedSourceRawSizeMetric() {
    final String readStep = "readFromKafka";
    final int numElements = 1000;
    final int numPartitionsPerTopic = 10;
    final int recordSize = 12; // The size of key and value is defined in ConsumerFactoryFn.

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

    KafkaIO.Read<byte[], Long> reader =
        KafkaIO.<byte[], Long>read()
            .withBootstrapServers("none")
            .withTopicPartitions(
                ImmutableList.of(new TopicPartition("test", 5), new TopicPartition("test", 8)))
            .withConsumerFactoryFn(
                new ConsumerFactoryFn(
                    topics, numPartitionsPerTopic, numElements, OffsetResetStrategy.EARLIEST))
            .withKeyDeserializer(ByteArrayDeserializer.class)
            .withValueDeserializer(LongDeserializer.class)
            .withMaxNumRecords(numElements / numPartitionsPerTopic * 2); // 2 is the # of partitions

    p.apply(readStep, reader.withoutMetadata()).apply(Values.create());

    PipelineResult result = p.run();

    MetricQueryResults metrics =
        result
            .metrics()
            .queryMetrics(
                MetricsFilter.builder()
                    .addNameFilter(
                        MetricNameFilter.inNamespace(KafkaUnboundedReader.METRIC_NAMESPACE))
                    .build());

    assertThat(
        metrics.getDistributions(),
        hasItem(
            attemptedMetricsResult(
                KafkaUnboundedReader.METRIC_NAMESPACE,
                KafkaUnboundedReader.RAW_SIZE_METRIC_PREFIX + "test-5",
                readStep,
                DistributionResult.create(
                    recordSize * numElements / numPartitionsPerTopic,
                    numElements / numPartitionsPerTopic,
                    recordSize,
                    recordSize))));

    assertThat(
        metrics.getDistributions(),
        hasItem(
            attemptedMetricsResult(
                KafkaUnboundedReader.METRIC_NAMESPACE,
                KafkaUnboundedReader.RAW_SIZE_METRIC_PREFIX + "test-8",
                readStep,
                DistributionResult.create(
                    recordSize * numElements / numPartitionsPerTopic,
                    numElements / numPartitionsPerTopic,
                    recordSize,
                    recordSize))));
  }