public static <K, V> Read<K, V> read() {
    return new AutoValue_CdapIO_Read.Builder<K, V>().build();
  }
@Test
  public void testReadFromCdapStreamingPlugin() {
    DirectOptions options = PipelineOptionsFactory.as(DirectOptions.class);
    options.setBlockOnRun(false);
    options.setRunner(DirectRunner.class);
    Pipeline p = Pipeline.create(options);

    EmployeeConfig pluginConfig =
        new ConfigWrapper<>(EmployeeConfig.class).withParams(TEST_EMPLOYEE_PARAMS_MAP).build();

    CdapIO.Read<String, String> read =
        CdapIO.<String, String>read()
            .withCdapPlugin(
                Plugin.createStreaming(
                    EmployeeStreamingSource.class,
                    Long::valueOf,
                    EmployeeReceiver.class,
                    config -> new Object[] {config}))
            .withPluginConfig(pluginConfig)
            .withKeyClass(String.class)
            .withValueClass(String.class)
            .withPullFrequencySec(PULL_FREQUENCY_SEC)
            .withStartOffset(START_OFFSET);

    List<String> storedRecords = EmployeeReceiver.getStoredRecords();

    PCollection<String> actual =
        p.apply("ReadStreamingTest", read)
            .setCoder(KvCoder.of(NullableCoder.of(StringUtf8Coder.of()), StringUtf8Coder.of()))
            .apply(Values.create());

    PAssert.that(actual).containsInAnyOrder(storedRecords);
    p.run().waitUntilFinish(Duration.standardSeconds(15));
  }