public static <K, V> Read<K, V> read() {
    return new AutoValue_CdapIO_Read.Builder<K, V>().build();
  }
@Test
  public void testReadBuildsCorrectly() {
    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);

    Plugin<String, String> cdapPlugin = read.getCdapPlugin();
    assertNotNull(cdapPlugin);
    assertEquals(EmployeeBatchSource.class, cdapPlugin.getPluginClass());
    assertEquals(EmployeeInputFormat.class, cdapPlugin.getFormatClass());
    assertEquals(EmployeeInputFormatProvider.class, cdapPlugin.getFormatProviderClass());
    assertNotNull(cdapPlugin.getContext());
    assertEquals(BatchSourceContextImpl.class, cdapPlugin.getContext().getClass());
    assertEquals(PluginConstants.PluginType.SOURCE, cdapPlugin.getPluginType());
    assertNotNull(cdapPlugin.getHadoopConfiguration());
    assertEquals(pluginConfig, read.getPluginConfig());
    assertEquals(String.class, read.getKeyClass());
    assertEquals(String.class, read.getValueClass());
  }