akka/alpakka-kafka

Hello I have situation where I lost some parts of messages after node restarts
last offset handling message in log 117246
after consumer restarted 117293
and there are no messages in range beetwen in logs

Here is my code
Please give me the hint where I am wrong
I need at least once guarantee

def start(...) = 
 for {
      streamCompletionRef                          = new AtomicReference[Option[Future[Done]]](None)
      assignedPartitionsRef                          = new AtomicReference(Set[TopicPartition]())
      consumer = makePartitionedConsumer(config, settings, controlRef, assignedPartitionsRef, handleMessage)
      _ <- Resource.make(startConsumerF(config, consumer, streamCompletionRef))(
            _ => shutdownConsumerF(name, controlRef, streamCompletionRef)
          )
    } yield ()


  private def startConsumerF(
      config: CommittableConsumerConfig,
      consumer: Source[Object, Unit],
      streamCompletionRef: AtomicReference[Option[Future[Done]]]
  )(implicit mat: Materializer): F[Unit] =
    F.delay {
      import config._
      RestartSource
        .onFailuresWithBackoff(minBackoff = backOffSettings.map(_.minBackoff).getOrElse(3).seconds,
                               maxBackoff = backOffSettings.map(_.maxBackoff).getOrElse(30).seconds,
                               randomFactor = backOffSettings.map(_.randomFactor).getOrElse(0.2))(() => consumer)
        .toMat(Sink.ignore) { case (_, streamCompletion) => streamCompletionRef.set(Some(streamCompletion)) }
        .run()
    }

  private def makePartitionedConsumer[K, V](
      config: CommittableConsumerConfig,
      consumerSettings: ConsumerSettings[K, V],
      controlRef: AtomicReference[Option[Consumer.Control]],
      assignedPartitionsRef: AtomicReference[Set[TopicPartition]],
      handleMessage: CommittableMessage[K, V] => F[Unit]
  )(implicit mat: Materializer, scheduler: Scheduler, logging: Logging[F]): Source[Done, Unit] = {
    import config._

    def handlePartitionSource(source: Source[CommittableMessage[K, V], NotUsed]): Future[Done] =
      source
        .mapAsync(maxPartitionParallelism.getOrElse(1)) { msg =>
          (logCommittableMessage(msg) >> handleMessage(msg))
            .as(msg.committableOffset)
            .runWithEmptyContext
            .runToFuture
        }
        .runWith(Committer.sink(CommitterSettings(config.committerConfig)))

    val subscriptions = Subscriptions
      .topics(topics.toSortedSet.toList: _*)
      .withPartitionAssignmentHandler(makePartitionAssignmentHandler(assignedPartitionsRef))

    Consumer
      .committablePartitionedSource(consumerSettings, subscriptions)
      .mapMaterializedValue(control => controlRef.set(Some(control)))
      .mapAsyncUnordered[Done](maxParallelism) { case (_, source) => handlePartitionSource(source) }
  }

  private def shutdownConsumerF(
      name: String,
      controlRef: AtomicReference[Option[Consumer.Control]],
      streamCompletionRef: AtomicReference[Option[Future[Done]]]
  )(implicit logger: Logging[F], exec: ExecutionContext): F[Unit] =
    for {
      _                     <- logger.info(s"Stopping Kafka consumer $name")
      maybeControl          <- F.delay(controlRef.get())
      maybeStreamCompletion <- F.delay(streamCompletionRef.get())
      _ <- maybeControl
            .map2(maybeStreamCompletion) { (control, streamCompletion) =>
              F.deferFuture(DrainingControl(control, streamCompletion).drainAndShutdown()).void
            }
            .getOrElse(F.unit)
            .flatMap(_ => logger.error(s"Successfully stopped Kafka consumer $name"))
            .onError[Unit, Throwable](e => logger.errorCause(s"Failed to stop Kafka consumer $name", e))
    } yield ()