public OffsetRange[] getNextOffsetRanges(Option<String> lastCheckpointStr, long sourceLimit, HoodieIngestionMetrics metrics) {
    // Come up with final set of OffsetRanges to read (account for new partitions, limit number of events)
    long maxEventsToReadFromKafka = getLongWithAltKeys(props, KafkaSourceConfig.MAX_EVENTS_FROM_KAFKA_SOURCE);

    long numEvents;
    if (sourceLimit == Long.MAX_VALUE) {
      numEvents = maxEventsToReadFromKafka;
      LOG.info("SourceLimit not configured, set numEvents to default value : {}", maxEventsToReadFromKafka);
    } else {
      numEvents = sourceLimit;
    }

    long minPartitions = getLongWithAltKeys(props, KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS);
    LOG.info("getNextOffsetRanges set config {} to {}", KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS.key(), minPartitions);

    return getNextOffsetRanges(lastCheckpointStr, numEvents, minPartitions, metrics);
  }
@Test
  public void testGetNextOffsetRangesFromSingleOffsetCheckpointNotApplicable() {
    testUtils.createTopic(testTopicName, 2);
    KafkaOffsetGen kafkaOffsetGen = new KafkaOffsetGen(getConsumerConfigs("latest", KAFKA_CHECKPOINT_TYPE_SINGLE_OFFSET));

    // incorrect number of partitions => exception (number of partitions is more than 1)
    String lastCheckpointString = "250";
    Exception exception = assertThrows(HoodieException.class,
        () -> kafkaOffsetGen.getNextOffsetRanges(Option.of(lastCheckpointString), 500, metrics));
    assertTrue(exception.getMessage().startsWith("Kafka topic " + testTopicName + " has 2 partitions (more than 1)"));
  }