# Prevent loss of first message on RestartSource.onFailuresWithBackoff with Transactional.source connected to Transaction.flow

**URL:** <https://discuss.akka.io/t/prevent-loss-of-first-message-on-restartsource-onfailureswithbackoff-with-transactional-source-connected-to-transaction-flow/6276>\
**Category:** Akka Libraries\
**Tags:** alpakka, reactive-kafka\
**Created:** [April 15, 2020, 10:00pm UTC](https://discuss.akka.io/t/prevent-loss-of-first-message-on-restartsource-onfailureswithbackoff-with-transactional-source-connected-to-transaction-flow/6276 "2020-04-15T22:00:39Z")\
**Posts on this page:** 1\
**Page:** 1

<div class="post-metadata">

**Author:** ![dineyw23](https://avatars.discourse-cdn.com/v4/letter/d/258eb7/32.png) [@dineyw23](https://discuss.akka.io/u/dineyw23)\
**Post date:** [April 15, 2020, 10:00pm UTC](https://discuss.akka.io/t/prevent-loss-of-first-message-on-restartsource-onfailureswithbackoff-with-transactional-source-connected-to-transaction-flow/6276/1 "2020-04-15T22:00:39Z")

</div>

I have a `Transactional.source` connected to `Transactional.flow` wrapped with `RestartSource.onFailuresWithBackoff`. I’m expecting the stream to restart itself in case of failures within my business logic. This works as I’m expecting when there is at least one committed message. The failed message is consumed again and reprocessed. But when my first message itself (First message in source topic ever) gives a failure in my business logic, the stream restarts once but the message is “lost”.  
What am I missing here or is this expected behavior?

I have following code from the documentation here : [https://doc.akka.io/docs/alpakka-kafka/current/transactions.html](https://doc.akka.io/docs/alpakka-kafka/current/transactions.html)

```auto
val stream = RestartSource.onFailuresWithBackoff(minBackoff = 1.seconds,maxBackoff = 30.seconds,randomFactor = 0.2)
 { () =>
  Transactional
    .source(consumerSettings, Subscriptions.topics(sourceTopic))
    .via(businessFlow())
    .map { msg =>
      ProducerMessage.single(new ProducerRecord(sinkTopic, msg.record.key, msg.record.value), msg.partitionOffset)
    }
    .via(Transactional.flow(producerSettings, transactionalId))
}

stream.runWith(Sink.ignore)

```

Thanks!