title: Gracefully restart a Reactive-Kafka Consumer Stream on failure
body: <p><strong>Problem</strong>
  When I restart/complete/STOP stream the old Consumer does not Die/Shutdown:</p>
  
  <pre><code>[INFO ] a.a.RepointableActorRef -
    Message [akka.kafka.KafkaConsumerActor$Internal$Stop$] 
    from Actor[akka://ufo-sightings/deadLetters]
    to Actor[akka://ufo-sightings/system/kafka-consumer-1#1896610594]
    was not delivered. [1] dead letters encountered.
  </code></pre>
  
  <p><strong>Description</strong>
  I'm building a service that receives a message from Kafka topic and sends the message to an external service via HTTP request.</p>
  
  <ol>
  <li><p>A connection with the external service can be broken, and my service needs to retry the request. </p></li>
  <li><p>Additionally, if there is an error in the Stream, entire stream needs to restart. </p></li>
  <li><p>Finally, sometimes I don't need the stream and its corresponding Kafka-consumer and I would like to shut down the entire stream</p></li>
  </ol>
  
  <p>So I have a Stream: </p>
  
  <pre><code>Consumer.committableSource(customizedSettings, subscriptions)
    .flatMapConcat(sourceFunction)
    .toMat(Sink.ignore)
    .run
  </code></pre>
  
  <p>Http request is sent in <code>sourceFunction</code></p>
  
  <p>I followed new Kafka Consumer Restart instructions in the new documentation</p>
  
  <pre><code>  RestartSource.withBackoff(
        minBackoff = 20.seconds,
        maxBackoff = 5.minutes,
        randomFactor = 0.2 ) { () =&gt;
            Consumer.committableSource(customizedSettings, subscriptions)
              .watchTermination() {
                  case (consumerControl, streamComplete) =&gt;
                    logger.info(s" Started Watching Kafka consumer id = ${consumer.id} termination: is shutdown: ${consumerControl.isShutdown}, is f completed: ${streamComplete.isCompleted}")
                    consumerControl.isShutdown.map(_ =&gt; logger.info(s"Shutdown of consumer finally happened id = ${consumer.id} at ${DateTime.now}"))
                    streamComplete
                      .flatMap { _ =&gt;
                        consumerControl.shutdown().map(_ -&gt; logger.info(s"3.consumer id = ${consumer.id} SHUTDOWN at ${DateTime.now} GRACEFULLY:CLOSED FROM UPSTREAM"))
                      }
                      .recoverWith {
                        case _ =&gt;
                          consumerControl.shutdown().map(_ -&gt; logger.info(s"3.consumer id = ${consumer.id} SHUTDOWN at ${DateTime.now} ERROR:CLOSED FROM UPSTREAM"))
                      }
               }
              .flatMapConcat(sourceFunction)
        }
        .viaMat(KillSwitches.single)(Keep.right)
        .toMat(Sink.ignore)(Keep.left)
        .run
  </code></pre>
  
  <p>There is an <a href="https://github.com/akka/reactive-kafka/issues/166" rel="nofollow noreferrer">issue</a> opened that discusses this non-terminating Consumer in a complex Akka-stream, but there is no solution yet.</p>
  
  <p>Is there a workaround that forces the Kafka Consumer termination</p>