public static <K, V> Write<K, V> write() {
    return new AutoValue_CdapIO_Write.Builder<K, V>().build();
  }
@Test
  public void testWriteWithCdapBatchSinkPlugin() throws IOException {
    List<KV<String, String>> data = new ArrayList<>();
    for (int i = 0; i < EmployeeInputFormat.NUM_OF_TEST_EMPLOYEE_RECORDS; i++) {
      data.add(KV.of(String.valueOf(i), EmployeeInputFormat.EMPLOYEE_NAME_PREFIX + i));
    }
    PCollection<KV<String, String>> input = p.apply(Create.of(data));

    EmployeeConfig pluginConfig =
        new ConfigWrapper<>(EmployeeConfig.class).withParams(TEST_EMPLOYEE_PARAMS_MAP).build();
    input.apply(
        "Write",
        CdapIO.<String, String>write()
            .withCdapPlugin(
                Plugin.createBatch(
                    EmployeeBatchSink.class,
                    EmployeeOutputFormat.class,
                    EmployeeOutputFormatProvider.class))
            .withPluginConfig(pluginConfig)
            .withKeyClass(String.class)
            .withValueClass(String.class)
            .withLocksDirPath(tmpFolder.getRoot().getAbsolutePath()));
    p.run();

    List<KV<String, String>> writtenOutput = EmployeeOutputFormat.getWrittenOutput();
    assertEquals(data.size(), writtenOutput.size());
    assertTrue(data.containsAll(writtenOutput));
    assertTrue(writtenOutput.containsAll(data));

    Mockito.verify(EmployeeOutputFormat.getOutputCommitter()).commitJob(Mockito.any());
  }