@GetInitialRestriction
  public OffsetRange getInitialRestriction(@Element PulsarSourceDescriptor pulsarSource) {
    long startTimestamp = 0L;
    long endTimestamp = Long.MAX_VALUE;

    if (pulsarSource.getStartOffset() != null) {
      startTimestamp = pulsarSource.getStartOffset();
    }

    if (pulsarSource.getEndOffset() != null) {
      endTimestamp = pulsarSource.getEndOffset();
    }

    return new OffsetRange(startTimestamp, endTimestamp);
  }
@Test
  public void testInitialRestrictionWithConsumerPosition() throws Exception {
    long expectedStartOffset = Instant.now().getMillis();
    OffsetRange result =
        dofnInstance.getInitialRestriction(
            PulsarSourceDescriptor.of(
                TOPIC, expectedStartOffset, null, null, SERVICE_URL, ADMIN_URL));
    assertEquals(new OffsetRange(expectedStartOffset, Long.MAX_VALUE), result);
  }