public static Read<String> read() {
    return new AutoValue_MongoDbGridFSIO_Read.Builder<String>()
        .setParser(TEXT_PARSER)
        .setCoder(StringUtf8Coder.of())
        .setConnectionConfiguration(ConnectionConfiguration.create())
        .setSkew(Duration.ZERO)
        .build();
  }
@Test
  public void testReadWithParser() {
    PCollection<KV<String, Integer>> output =
        pipeline.apply(
            MongoDbGridFSIO.read()
                .withUri("mongodb://localhost:" + port)
                .withDatabase(DATABASE)
                .withBucket("mapBucket")
                .<KV<String, Integer>>withParser(
                    (input, callback) -> {
                      try (final BufferedReader reader =
                          new BufferedReader(
                              new InputStreamReader(
                                  input.getInputStream(), StandardCharsets.UTF_8))) {
                        String line = reader.readLine();
                        while (line != null) {
                          try (Scanner scanner = new Scanner(line.trim())) {
                            scanner.useDelimiter("\\t");
                            long timestamp = scanner.nextLong();
                            String name = scanner.next();
                            int score = scanner.nextInt();
                            callback.output(KV.of(name, score), new Instant(timestamp));
                          }
                          line = reader.readLine();
                        }
                      }
                    })
                .withSkew(Duration.millis(3610000L))
                .withCoder(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of())));

    PAssert.thatSingleton(output.apply("Count All", Count.globally())).isEqualTo(50100L);

    PAssert.that(output.apply("Max PerElement", Max.integersPerKey()))
        .satisfies(
            input -> {
              for (KV<String, Integer> element : input) {
                assertEquals(101, element.getValue().longValue());
              }
              return null;
            });

    pipeline.run();
  }