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 testUnboundedSourceWithPartiallyMatchedPattern() {
    int numElements = 1000;
    long numMatchedElements = numElements / 2; // Expected elements if split across 2 topics

    List<String> topics = ImmutableList.of("test", "Test");
    String bootStrapServer = "none";

    KafkaIO.Read<byte[], Long> reader =
        KafkaIO.<byte[], Long>read()
            .withBootstrapServers(bootStrapServer)
            .withTopicPattern("[a-z]est")
            .withConsumerFactoryFn(
                new ConsumerFactoryFn(topics, 1, numElements, OffsetResetStrategy.EARLIEST))
            .withKeyDeserializer(ByteArrayDeserializer.class)
            .withValueDeserializer(LongDeserializer.class)
            .withMaxNumRecords(numMatchedElements);

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

    // With 1 partition per topic element to partition allocation alternates between test and Test,
    // producing even elements for test and odd elements for Test.
    // The pattern only matches test, so we expect even elements.
    PAssert.that(input).satisfies(new AssertMultipleOf(2));

    PAssert.thatSingleton(input.apply("Count", Count.globally())).isEqualTo(numMatchedElements);

    PipelineResult result = p.run();
    assertThat(
        Lineage.query(result.metrics(), Lineage.Type.SOURCE),
        hasItem(String.format("kafka:%s.test", bootStrapServer)));
    assertThat(
        Lineage.query(result.metrics(), Lineage.Type.SOURCE),
        not(hasItem(String.format("kafka:%s.Test", bootStrapServer))));
  }