@GetInitialRestriction
  public OffsetRange initialRestriction(@Element KafkaSourceDescriptor kafkaSourceDescriptor) {
    Map<String, Object> updatedConsumerConfig =
        overrideBootstrapServersConfig(consumerConfig, kafkaSourceDescriptor);
    TopicPartition partition = kafkaSourceDescriptor.getTopicPartition();
    LOG.info("Creating Kafka consumer for initial restriction for {}", partition);
    try (Consumer<byte[], byte[]> offsetConsumer = consumerFactoryFn.apply(updatedConsumerConfig)) {
      ConsumerSpEL.evaluateAssign(offsetConsumer, ImmutableList.of(partition));
      long startOffset;
      @Nullable Instant startReadTime = kafkaSourceDescriptor.getStartReadTime();
      if (kafkaSourceDescriptor.getStartReadOffset() != null) {
        startOffset = kafkaSourceDescriptor.getStartReadOffset();
      } else if (startReadTime != null) {
        startOffset = ConsumerSpEL.offsetForTime(offsetConsumer, partition, startReadTime);
      } else {
        startOffset = offsetConsumer.position(partition);
      }

      long endOffset = Long.MAX_VALUE;
      @Nullable Instant stopReadTime = kafkaSourceDescriptor.getStopReadTime();
      if (kafkaSourceDescriptor.getStopReadOffset() != null) {
        endOffset = kafkaSourceDescriptor.getStopReadOffset();
      } else if (stopReadTime != null) {
        endOffset = ConsumerSpEL.offsetForTime(offsetConsumer, partition, stopReadTime);
      }
      new OffsetRange(startOffset, endOffset);
      Lineage.getSources()
          .add(
              "kafka",
              ImmutableList.of(
                  (String) updatedConsumerConfig.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
                  MoreObjects.firstNonNull(kafkaSourceDescriptor.getTopic(), partition.topic())));
      return new OffsetRange(startOffset, endOffset);
    }
  }
@Test
  public void testInitialRestrictionWhenHasStopOffset() throws Exception {
    long expectedStartOffset = 10L;
    long expectedStopOffset = 20L;
    consumer.setStartOffsetForTime(15L, Instant.now());
    consumer.setStopOffsetForTime(18L, Instant.now());
    consumer.setCurrentPos(5L);
    OffsetRange result =
        dofnInstance.initialRestriction(
            KafkaSourceDescriptor.of(
                topicPartition,
                expectedStartOffset,
                null,
                expectedStopOffset,
                null,
                ImmutableList.of()));
    assertEquals(new OffsetRange(expectedStartOffset, expectedStopOffset), result);
  }