@ProcessElement
  public ProcessContinuation processElement(
      @Element PulsarSourceDescriptor pulsarSourceDescriptor,
      RestrictionTracker<OffsetRange, Long> tracker,
      WatermarkEstimator watermarkEstimator,
      OutputReceiver<PulsarMessage> output)
      throws IOException {
    long startTimestamp = tracker.currentRestriction().getFrom();
    String topicDescriptor = pulsarSourceDescriptor.getTopic();
    try (Reader<byte[]> reader = newReader(this.client, topicDescriptor)) {
      if (startTimestamp > 0) {
        reader.seek(startTimestamp);
      }
      while (true) {
        if (reader.hasReachedEndOfTopic()) {
          reader.close();
          return ProcessContinuation.stop();
        }
        Message<byte[]> message = reader.readNext();
        if (message == null) {
          return ProcessContinuation.resume();
        }
        Long currentTimestamp = message.getPublishTime();
        // if tracker.tryclaim() return true, sdf must execute work otherwise
        // doFn must exit processElement() without doing any work associated
        // or claiming more work
        if (!tracker.tryClaim(currentTimestamp)) {
          reader.close();
          return ProcessContinuation.stop();
        }
        if (pulsarSourceDescriptor.getEndMessageId() != null) {
          MessageId currentMsgId = message.getMessageId();
          boolean hasReachedEndMessageId =
              currentMsgId.compareTo(pulsarSourceDescriptor.getEndMessageId()) == 0;
          if (hasReachedEndMessageId) {
            return ProcessContinuation.stop();
          }
        }
        PulsarMessage pulsarMessage =
            new PulsarMessage(message.getTopicName(), message.getPublishTime(), message);
        Instant outputTimestamp = extractOutputTimestampFn.apply(message);
        output.outputWithTimestamp(pulsarMessage, outputTimestamp);
      }
    }
  }
@Test
  public void testProcessElementWithEmptyRecords() throws Exception {
    MockOutputReceiver receiver = new MockOutputReceiver();
    fakePulsarReader.emptyMockRecords();
    OffsetRangeTracker tracker = new OffsetRangeTracker(new OffsetRange(0L, Long.MAX_VALUE));
    DoFn.ProcessContinuation result =
        dofnInstance.processElement(
            PulsarSourceDescriptor.of(TOPIC, null, null, null, SERVICE_URL, ADMIN_URL),
            tracker,
            null,
            (DoFn.OutputReceiver) receiver);
    assertEquals(DoFn.ProcessContinuation.resume(), result);
    assertTrue(receiver.getOutputs().isEmpty());
  }