public static <K, V> Read<K, V> read() {
    return new AutoValue_CdapIO_Read.Builder<K, V>().build();
  }
@Test
  public void testReadFromCdapBatchPlugin() {
    EmployeeConfig pluginConfig =
        new ConfigWrapper<>(EmployeeConfig.class).withParams(TEST_EMPLOYEE_PARAMS_MAP).build();
    CdapIO.Read<String, String> read =
        CdapIO.<String, String>read()
            .withCdapPlugin(
                Plugin.createBatch(
                    EmployeeBatchSource.class,
                    EmployeeInputFormat.class,
                    EmployeeInputFormatProvider.class))
            .withPluginConfig(pluginConfig)
            .withKeyClass(String.class)
            .withValueClass(String.class);

    List<KV<String, String>> expected = new ArrayList<>();
    for (int i = 1; i < EmployeeInputFormat.NUM_OF_TEST_EMPLOYEE_RECORDS; i++) {
      expected.add(KV.of(String.valueOf(i), EmployeeInputFormat.EMPLOYEE_NAME_PREFIX + i));
    }
    PCollection<KV<String, String>> actual = p.apply("ReadBatchTest", read);
    PAssert.that(actual).containsInAnyOrder(expected);
    p.run();
  }