@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 testInitialRestrictionWithConsumerEndPosition() throws Exception {
    long startOffset = fakePulsarReader.getStartTimestamp();
    long endOffset = fakePulsarReader.getEndTimestamp();
    OffsetRange result =
        dofnInstance.getInitialRestriction(
            PulsarSourceDescriptor.of(TOPIC, startOffset, endOffset, null, SERVICE_URL, ADMIN_URL));
    assertEquals(new OffsetRange(startOffset, endOffset), result);
  }