private def adjustStartingOffset()

in core/src/main/scala/org/apache/spark/sql/eventhubs/EventHubsSource.scala [246:260]

15 lines of code
3 McCabe index (conditional complexity)


  private def adjustStartingOffset(
      from: Map[NameAndPartition, SequenceNumber]): Map[NameAndPartition, SequenceNumber] = {
    from.map {
      case (nAndP, seqNo) =>
        if (seqNo < earliestSeqNos.get(nAndP)) {
          reportDataLoss(
            s"Starting seqNo $seqNo in partition ${nAndP.partitionId} of EventHub ${nAndP.ehName} " +
              s"is behind the earliest sequence number ${earliestSeqNos.get(nAndP)} " +
              s"present in the service. Some events may have expired and been missed.")
          nAndP -> earliestSeqNos.get(nAndP)
        } else {
          nAndP -> seqNo
        }
    }
  }