public static Read read() {
    return new AutoValue_PulsarIO_Read.Builder()
        .setPulsarClient(PulsarIOUtils.PULSAR_CLIENT_SERIALIZABLE_FUNCTION)
        .build();
  }
@Test
  public void testReadFromSimpleTopic() {
    try {
      List<PulsarMessage> inputsMock = produceMessages();
      PulsarIO.Read reader =
          PulsarIO.read()
              .withClientUrl(pulsarContainer.getPulsarBrokerUrl())
              .withAdminUrl(pulsarContainer.getHttpServiceUrl())
              .withTopic(TOPIC)
              .withStartTimestamp(startTime)
              .withEndTimestamp(endExpectedTime)
              .withPublishTime();
      testPipeline.apply(reader).apply(ParDo.of(new PulsarRecordsMetric()));

      PipelineResult pipelineResult = testPipeline.run();
      MetricQueryResults metrics =
          pipelineResult
              .metrics()
              .queryMetrics(
                  MetricsFilter.builder()
                      .addNameFilter(
                          MetricNameFilter.named(
                              PulsarIOTest.class.getName(), "PulsarRecordsCounter"))
                      .build());
      long recordsCount = 0;
      for (MetricResult<Long> metric : metrics.getCounters()) {
        if (metric
            .getName()
            .toString()
            .equals("org.apache.beam.sdk.io.pulsar.PulsarIOTest:PulsarRecordsCounter")) {
          recordsCount = metric.getAttempted();
          break;
        }
      }
      assertEquals(inputsMock.size(), (int) recordsCount);

    } catch (PulsarClientException e) {
      LOG.error(e.getMessage());
    }
  }