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 testReadAvroGenericRecordsWithConfluentSchemaRegistry() {
    int numElements = 100;
    String topic = "my_topic";
    String schemaRegistryUrl = "mock://my-scope-name";
    String keySchemaSubject = topic + "-key";
    String valueSchemaSubject = topic + "-value";

    List<KV<GenericRecord, GenericRecord>> inputs = new ArrayList<>();
    for (int i = 0; i < numElements; i++) {
      inputs.add(
          KV.of(
              new AvroGeneratedUser("KeyName" + i, i, "color" + i),
              new AvroGeneratedUser("ValueName" + i, i, "color" + i)));
    }

    KafkaIO.Read<GenericRecord, GenericRecord> reader =
        KafkaIO.<GenericRecord, GenericRecord>read()
            .withBootstrapServers("localhost:9092")
            .withTopic(topic)
            .withKeyDeserializer(
                mockDeserializerProvider(schemaRegistryUrl, keySchemaSubject, null))
            .withValueDeserializer(
                mockDeserializerProvider(schemaRegistryUrl, valueSchemaSubject, null))
            .withConsumerFactoryFn(
                new ConsumerFactoryFn(
                    ImmutableList.of(topic),
                    1,
                    numElements,
                    OffsetResetStrategy.EARLIEST,
                    new KeyAvroSerializableFunction(topic, schemaRegistryUrl),
                    new ValueAvroSerializableFunction(topic, schemaRegistryUrl)))
            .withMaxNumRecords(numElements);

    PCollection<KV<GenericRecord, GenericRecord>> input = p.apply(reader.withoutMetadata());

    PAssert.that(input).containsInAnyOrder(inputs);
    p.run();
  }