public static <K, V> Write<K, V> write() {
    return new AutoValue_KafkaIO_Write.Builder<K, V>()
        .setWriteRecordsTransform(writeRecords())
        .build();
  }
@Test
  public void testExactlyOnceSinkWithSendException() throws Throwable {

    if (!ProducerSpEL.supportsTransactions()) {
      LOG.warn(
          "testExactlyOnceSink() is disabled as Kafka client version does not support transactions.");
      return;
    }

    thrown.expect(KafkaException.class);
    thrown.expectMessage("fakeException");

    String topic = "test";

    p.apply(Create.of(ImmutableList.of(KV.of(1, 1L), KV.of(2, 2L))))
        .apply(
            KafkaIO.<Integer, Long>write()
                .withBootstrapServers("none")
                .withTopic(topic)
                .withKeySerializer(IntegerSerializer.class)
                .withValueSerializer(LongSerializer.class)
                .withEOS(1, "testException")
                .withConsumerFactoryFn(
                    new ConsumerFactoryFn(
                        Lists.newArrayList(topic), 10, 10, OffsetResetStrategy.EARLIEST))
                .withProducerFactoryFn(new SendErrorProducerFactory()));

    try {
      p.run();
    } catch (PipelineExecutionException e) {
      // throwing inner exception helps assert that first exception is thrown from the Sink
      throw e.getCause();
    }
  }