public static <K, V> WriteRecords<K, V> writeRecords() {
    return new AutoValue_KafkaIO_WriteRecords.Builder<K, V>()
        .setProducerConfig(WriteRecords.DEFAULT_PRODUCER_PROPERTIES)
        .setEOS(false)
        .setNumShards(0)
        .setConsumerFactoryFn(KafkaIOUtils.KAFKA_CONSUMER_FACTORY_FN)
        .setBadRecordRouter(BadRecordRouter.THROWING_ROUTER)
        .setBadRecordErrorHandler(new DefaultErrorHandler<>())
        .build();
  }
@Test
  public void testRecordsSink() throws Exception {
    // Simply read from kafka source and write to kafka sink using ProducerRecord transform. Then
    // verify the records are correctly published to mock kafka producer.

    int numElements = 1000;

    try (MockProducerWrapper producerWrapper = new MockProducerWrapper(new LongSerializer())) {

      ProducerSendCompletionThread completionThread =
          new ProducerSendCompletionThread(producerWrapper.mockProducer).start();

      String topic = "test";

      p.apply(mkKafkaReadTransform(numElements, new ValueAsTimestampFn()).withoutMetadata())
          .apply(ParDo.of(new KV2ProducerRecord(topic)))
          .setCoder(ProducerRecordCoder.of(VarIntCoder.of(), VarLongCoder.of()))
          .apply(
              KafkaIO.<Integer, Long>writeRecords()
                  .withBootstrapServers("none")
                  .withTopic(topic)
                  .withKeySerializer(IntegerSerializer.class)
                  .withValueSerializer(LongSerializer.class)
                  .withInputTimestamp()
                  .withProducerFactoryFn(new ProducerFactoryFn(producerWrapper.producerKey)));

      p.run();

      completionThread.shutdown();

      verifyProducerRecords(producerWrapper.mockProducer, topic, numElements, false, true);
    }
  }