# Kafka Draining Control with restart mechanism

**URL:** <https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855>\
**Category:** Akka Streams & Alpakka\
**Tags:** kafka, alpakka\
**Created:** [April 4, 2019, 7:51am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855 "2019-04-04T07:51:59Z")\
**Posts on this page:** 12\
**Page:** 1

<div class="post-metadata">

**Author:** ![sigurd](https://avatars.discourse-cdn.com/v4/letter/s/e95f7d/32.png) [@sigurd](https://discuss.akka.io/u/sigurd)\
**Post date:** [April 4, 2019, 7:51am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/1 "2019-04-04T07:51:59Z")

</div>

Hey guys,

I’m working with alpakka (slick,cassandra,kafka).  
For consuming a Kafka topic I use the Draining Control combined with the new Committer Sink.  
[https://doc.akka.io/docs/alpakka-kafka/current/consumer.html#committer-sink](https://doc.akka.io/docs/alpakka-kafka/current/consumer.html#committer-sink)

Thats nice, but I’m unsure how I can now react (restart) on errors that probably after not existing anymore after a restart (connection problems, …).  
I would like to use the RestartSource but a DrainingControl is not a Source or CompletionStage.

Whats the trick? 😃

Best  
Sigurd

---

<div class="post-metadata">

**Author:** ![ennru](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/ennru/32/1495_2.png) [@ennru](https://discuss.akka.io/u/ennru)\
**Post date:** [April 25, 2019, 2:26pm UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/2 "2019-04-25T14:26:36Z")

</div>

Hi Sigurd,

Sorry for not answering earlier.

You’re right, there is no good way to get the materialized value out of the `RestartSource`. One way of getting access to the latest materialized control is to have an `AtomicReference` which is populated from the inner stream creation logic in `mapMaterializedValue` as shown in the docs: [https://doc.akka.io/docs/alpakka-kafka/current/errorhandling.html](https://doc.akka.io/docs/alpakka-kafka/current/errorhandling.html)

I hope that helps,  
Enno.

---

<div class="post-metadata">

**Author:** ![sigurd](https://avatars.discourse-cdn.com/v4/letter/s/e95f7d/32.png) [@sigurd](https://discuss.akka.io/u/sigurd)\
**Post date:** [April 25, 2019, 2:44pm UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/3 "2019-04-25T14:44:29Z")

</div>

Thank you for that hint, thats interesting.

I currently use a decider function to restart the source.

```auto
akka.japi.function.Function<Throwable, Supervision.Directive> decider = exc -> {
            if (exc instanceof CoordinatorNotAvailableException) {
                return Supervision.restart();
            }
            else if (exc instanceof RetriableException){
                return Supervision.restart();
            }
            else if (exc instanceof java.util.concurrent.TimeoutException){
                return Supervision.restart();
            }
            else{
                return Supervision.stop();
            }
        };

```

And additionally I add a shut down hook with a CoordinatedShutdown which terminates the Draining Control in phase service-unbind (with phase default configuration).  
I’m little bit unsure which phase is the best. I get now failed graph stages on shut down.

---

<div class="post-metadata">

**Author:** ![nur](https://avatars.discourse-cdn.com/v4/letter/n/e9c0ed/32.png) [@nur](https://discuss.akka.io/u/nur)\
**Post date:** [November 7, 2019, 9:28am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/4 "2019-11-07T09:28:30Z")

</div>

Hi Enno,

I am new with Alpakka. And I am trying to use RestartSource with DrainingControl.apply. But I get a compile error below. I can do away with that line but I’m worried that my flow will not drain properly. Basically my flow is a consumer that goes to a producer. How can I ensure that the flow will drain but wrapping it in a RestartSource?

error [error] found : akka.stream.scaladsl.RunnableGraph[akka.kafka.scaladsl.Consumer.DrainingControl[akka.Done]]  
[error] required: akka.stream.scaladsl.Source[?, \_]  
[error] .mapMaterializedValue(DrainingControl.apply)

---

<div class="post-metadata">

**Author:** ![ennru](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/ennru/32/1495_2.png) [@ennru](https://discuss.akka.io/u/ennru)\
**Post date:** [November 7, 2019, 2:08pm UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/5 "2019-11-07T14:08:00Z")

</div>

Hi @nur

Are you sure you need a `RestartSource`? For most use cases it is not necessary.

When using `RestartSource` you would need to create the `DrainingControl` more explicitly as the stream doesn’t get access to the “latest” materialized value from the consumer source.  
You need to pass in the value from the `AtomicReference` and the sink’s stream completion future to create the `DrainingControl`.

Cheers,  
Enno.

---

<div class="post-metadata">

**Author:** ![nur](https://avatars.discourse-cdn.com/v4/letter/n/e9c0ed/32.png) [@nur](https://discuss.akka.io/u/nur)\
**Post date:** [November 8, 2019, 2:45am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/6 "2019-11-08T02:45:59Z")

</div>

Hi @Enno,

I have some business logic in between the Consumer and Producer stage that might cause an exception. I like to automatically restart the flow with a backoff if that happens. My output is to the Producer so would I still need a materialized value?

I assumed that I still needed it and I did what you recommended and it compiled. Thanks a lot for your help. I still have an issue with getting results for unit tests using testcontainers. But I will post it as a separate ticket. Thanks again.

Regards,

---

<div class="post-metadata">

**Author:** ![NullPointerException](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/nullpointerexception/32/1741_2.png) [@NullPointerException](https://discuss.akka.io/u/NullPointerException)\
**Post date:** [September 3, 2020, 2:35pm UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/7 "2020-09-03T14:35:40Z")

</div>

Are you looking for something like this?

```auto
  val control = new AtomicReference[Consumer.Control](Consumer.NoopControl)

  val result: Future[Foo] = RestartSource
    .onFailuresWithBackoff(
      minBackoff = 3.seconds,
      maxBackoff = 30.seconds,
      randomFactor = 0.2
    ) { () =>
      \\ you source here
    }
    .runWith(dbSink)

  val drainingControl = DrainingControl(Tuple2(control.get(), result))

```

---

<div class="post-metadata">

**Author:** ![nur](https://avatars.discourse-cdn.com/v4/letter/n/e9c0ed/32.png) [@nur](https://discuss.akka.io/u/nur)\
**Post date:** [September 7, 2020, 3:04am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/8 "2020-09-07T03:04:52Z")

</div>

> [@NullPointerException](#):
>
> `control`

I use a committable source. So this is what I did.

```auto
restartSource
      .toMat({
        Committer.sink(committerSettings)
      })(
        Keep.right
      )
      .mapMaterializedValue(f => {
        DrainingControl.apply(Tuple2(control.get(), f))
      })
      .run()
```

---

<div class="post-metadata">

**Author:** ![ennru](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/ennru/32/1495_2.png) [@ennru](https://discuss.akka.io/u/ennru)\
**Post date:** [September 11, 2020, 9:25am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/9 "2020-09-11T09:25:41Z")

</div>

You need to construct the `DrainingControl` with the latest `Control` when you want to shut down the stream.  
Passing in `Consumer.NoopControl` as @NullPointerException suggested is useful so there is a valid value until the consumer is established.

But don’t create the `DrainingControl` immediately after creating the stream as that will most likely catch the `NoopControl`. Instead, you may use `Control.drainAndShutdown` and pass in the stream completion signal to it.

Cheers,  
Enno.

---

<div class="post-metadata">

**Author:** ![nur](https://avatars.discourse-cdn.com/v4/letter/n/e9c0ed/32.png) [@nur](https://discuss.akka.io/u/nur)\
**Post date:** [September 14, 2020, 10:14am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/10 "2020-09-14T10:14:25Z")

</div>

Hi @ennru,

Am I correct in using RestartSource? I use explicit commit of offset as I am doing something with the message from the kafka source. I want to repeat in case something fails. Will my code be re-triggerred if I fail to commit the offset, even if I don’t use RestartSource?

If I am correct in using RestartSource, isn’t mapMaterialized the place where I should create the DrainingControl? I am not sure how to pass the stream completion. Wouldn’t calling drainAndShutdown() in the code below work?

```auto
    val control = new AtomicReference[Control](Consumer.NoopControl)

    val restartSource = 
      RestartSource
        .onFailuresWithBackoff() { 
          // logic here
        }
    

    val drainingControl = restartSource
      .toMat({
        Committer.sink(openedCommitterSettings)
      })(
        Keep.right
      )
      .mapMaterializedValue(f => {
        DrainingControl.apply(Tuple2(control.get(), f))
      })
      .run()

    drainingControl.drainAndShutdown()
```

---

<div class="post-metadata">

**Author:** ![nur](https://avatars.discourse-cdn.com/v4/letter/n/e9c0ed/32.png) [@nur](https://discuss.akka.io/u/nur)\
**Post date:** [April 6, 2021, 5:12am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/11 "2021-04-06T05:12:26Z")

</div>

Hi @ennru,

I haven’t gotten a reply from you. I experience a very rare problem with my implementation that I can’t pinpoint the cause. So I returned to this thread. Is this what you are recommending?

val drainingControl = restartSource  
.toMat({  
Committer.sink(openedCommitterSettings)  
})(  
Keep.right  
)  
.mapMaterializedValue(f =\> {  
control.get(f).drainAndShutdown)  
})  
.run()

Regards,

---

<div class="post-metadata">

**Author:** ![kafkauser](https://avatars.discourse-cdn.com/v4/letter/k/f4b2a3/32.png) [@kafkauser](https://discuss.akka.io/u/kafkauser)\
**Post date:** [March 10, 2022, 6:22am UTC](https://discuss.akka.io/t/kafka-draining-control-with-restart-mechanism/3855/13 "2022-03-10T06:22:03Z")

</div>

hi @ennru / @nur  
I am facing class cast exception when using the above solution

streamCompletion =  
(CompletionStage) RestartSource.onFailuresWithBackoff()

Caused by: java.lang.ClassCastException: akka.NotUsed$ cannot be cast to java.util.concurrent.CompletionStage

If i am not casting then facing issue at compile time.  
Please share the correct example to use.