title: How to shutdown a Kafka ConsumerConnector
body: <p>I have a system that pulls messages from a Kafka topic, and when it's unable to process messages because some external resource is unavailable, it shuts down the consumer, returns the message to the topic, and waits some time before starting the consumer again. The only problem is, shutting down doesn't work. Here's what I see in my logs:</p>
  
  <blockquote>
    <p>2014-09-30 08:24:10,918 - com.example.kafka.KafkaConsumer [info] - [application-akka.actor.workflow-context-8] Shutting down kafka consumer for topic new-problem-reports
    2014-09-30 08:24:10,927 - clients.kafka.ProblemReportObserver [info] - [application-akka.actor.workflow-context-8] Consumer shutdown
    2014-09-30 08:24:11,946 - clients.kafka.ProblemReportObserver [warn] - [application-akka.actor.workflow-context-8] Sending 7410-1412090624000 back to the queue
    2014-09-30 08:24:12,021 - clients.kafka.ProblemReportObserver [debug] - [kafka-akka.actor.kafka-consumer-worker-context-9] Message from partition 0: key=7410-1412090624000, msg=7410-1412090624000</p>
  </blockquote>
  
  <p>There's a few layers at work here, but the important code is:</p>
  
  <p>In <code>KafkaConsumer.scala</code>:</p>
  
  <pre><code>protected def consumer: ConsumerConnector = Consumer.create(config.asKafkaConfig)
  def shutdown() = {
    logger.info(s"Shutting down kafka consumer for topic ${config.topic}")
    consumer.shutdown()
  }
  </code></pre>
  
  <p>In the routine that observes messages:</p>
  
  <pre><code>(processor ? ProblemReportRequest(problemReportKey)).map {
    case e: ConnectivityInterruption =&gt;
      val backoff = 10.seconds
      logger.warn(s"Can't connect to essential services, pausing for $backoff", e)
      stop()
      // XXX: Shutdown isn't instantaneous, so returning has to happen after a delay.
      // Unfortunately, there's still a race condition here, plus there's a chance the
      // system will be shut down before the message has been returned.
      system.scheduler.scheduleOnce(100 millis) { returnMessage(message) }
      system.scheduler.scheduleOnce(backoff) { start() }
      false
    case e: Exception =&gt; returnMessage(message, e)
    case _ =&gt; true
  }.recover { case e =&gt; returnMessage(message, e) }
  </code></pre>
  
  <p>And the stop method:</p>
  
  <pre><code>def stop() = {
    if (consumerRunning.get()) {
      consumer.shutdown()
      consumerRunning.compareAndSet(true, false)
      logger.info("Consumer shutdown")
    } else {
      logger.info("Consumer is already shutdown")
    }
    !consumerRunning.get()
  }
  </code></pre>
  
  <p>Is this a bug, or am I doing it wrong?</p>