Alpakka Kafka connector (akka-stream-kafka) example

Simple solution to use Alpakka Kafka connector to produce and consume kafka messages.

I assume that you have 2 scala apps, a producer and a consumer.

Producer

Add the following dependencies:

• "com.typesafe.akka" %% "akka-stream-kafka" % "0.21.1"

Create an application.conf:

# Properties for akka.kafka.ProducerSettings can be
# defined in this section or a configuration section with
# the same layout.
akka.kafka.producer {
  # Tuning parameter of how many sends that can run in parallel.
  # parallelism = 100

  # How long to wait for `KafkaProducer.close`
  # close-timeout = 60s

  # Fully qualified config path which holds the dispatcher configuration
  # to be used by the producer stages. Some blocking may occur.
  # When this value is empty, the dispatcher configured for the stream
  # will be used.
  # use-dispatcher = "akka.kafka.default-dispatcher"

  # The time interval to commit a transaction when using the `Transactional.sink` or `Transactional.flow`
  # eos-commit-interval = 100ms

  # Properties defined by org.apache.kafka.clients.producer.ProducerConfig
  # can be defined in this configuration section.
  kafka-clients {
    bootstrap.servers = "quickstart.cloudera:9092"
  }
}

Create a producer.scala:

package fakeProducer

import akka.Done
import akka.actor.ActorSystem
import akka.kafka.ProducerSettings
import akka.kafka.scaladsl.Producer
import akka.stream.{ActorMaterializer, Materializer}
import akka.stream.scaladsl.Source
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer

import scala.concurrent.{ExecutionContextExecutor, Future}
import scala.util.{Failure, Success}

object App {
  def main(args: Array[String]): Unit = {
    println("Hello from producer")

    implicit val system: ActorSystem = ActorSystem("producer-sample")
    implicit val materializer: Materializer = ActorMaterializer()

    val producerSettings =
      ProducerSettings(system, new StringSerializer, new StringSerializer)

    val done: Future[Done] =
      Source(1 to 100)
        .map(value => new ProducerRecord[String, String]("test-topic", "msg " + value))
        .runWith(Producer.plainSink(producerSettings))

    implicit val ec: ExecutionContextExecutor = system.dispatcher
    done onComplete  {
      case Success(_) => println("Done"); system.terminate()
      case Failure(err) => println(err.toString); system.terminate()
    }
  }
}

Consumer

Add the following dependencies:

• "com.typesafe.akka" %% "akka-stream-kafka" % "0.21.1"

Create an application.conf file:

# Properties for akka.kafka.ConsumerSettings can be
# defined in this section or a configuration section with
# the same layout.
akka.kafka.consumer {
  # Tuning property of scheduled polls.
  # poll-interval = 50ms

  # Tuning property of the `KafkaConsumer.poll` parameter.
  # Note that non-zero value means that the thread that
  # is executing the stage will be blocked.
  # poll-timeout = 50ms

  # The stage will await outstanding offset commit requests before
  # shutting down, but if that takes longer than this timeout it will
  # stop forcefully.
  # stop-timeout = 30s

  # How long to wait for `KafkaConsumer.close`
  # close-timeout = 20s

  # If offset commit requests are not completed within this timeout
  # the returned Future is completed `CommitTimeoutException`.
  # commit-timeout = 15s

  # If commits take longer than this time a warning is logged
  # commit-time-warning = 1s

  # If for any reason `KafkaConsumer.poll` blocks for longer than the configured
  # poll-timeout then it is forcefully woken up with `KafkaConsumer.wakeup`.
  # The KafkaConsumerActor will throw
  # `org.apache.kafka.common.errors.WakeupException` which will be ignored
  # until `max-wakeups` limit gets exceeded.
  # wakeup-timeout = 60s

  # After exceeding maxinum wakeups the consumer will stop and the stage and fail.
  # Setting it to 0 will let it ignore the wakeups and try to get the polling done forever.
  # max-wakeups = 10

  # If set to a finite duration, the consumer will re-send the last committed offsets periodically
  # for all assigned partitions. See https://issues.apache.org/jira/browse/KAFKA-4682.
  # commit-refresh-interval = infinite

  # If enabled, log stack traces before waking up the KafkaConsumer to give
  # some indication why the KafkaConsumer is not honouring the `poll-timeout`
  #wakeup-debug = true

  # Fully qualified config path which holds the dispatcher configuration
  # to be used by the KafkaConsumerActor. Some blocking may occur.
  # use-dispatcher = "akka.kafka.default-dispatcher"

  # Properties defined by org.apache.kafka.clients.consumer.ConsumerConfig
  # can be defined in this configuration section.
  kafka-clients {
    # auto-commit disabled by default
    # Setting enable.auto.commit means that offsets are committed automatically 
    #  with a frequency controlled by the config auto.commit.interval.ms.
    enable.auto.commit = true

    bootstrap.servers = "quickstart.cloudera:9092"
    group.id = "test-group1"

    auto.offset.reset = "earliest"
  }

  # Time to wait for pending requests when a partition is closed
  # wait-close-partition = 500ms
}

Create a consumer.scala file:

package fakeConsumer

import akka.actor.ActorSystem
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.kafka.scaladsl.Consumer
import akka.stream.scaladsl.Sink
import akka.stream.{ActorMaterializer, Materializer}
import org.apache.kafka.common.serialization.StringDeserializer

import scala.concurrent.ExecutionContextExecutor
import scala.util.{Failure, Success}

object App {
  def main(args: Array[String]): Unit = {
    println("Hello from hBaseWriter")

    implicit val system: ActorSystem = ActorSystem("consumer-sample")
    implicit val materializer: Materializer = ActorMaterializer()

    val consumerSettings =
      ConsumerSettings(system, new StringDeserializer, new StringDeserializer)

    val done = Consumer
      .plainSource(consumerSettings, Subscriptions.topics("test-topic"))
      .runWith(Sink.foreach(println)) // just print each message for debugging

    implicit val ec: ExecutionContextExecutor = system.dispatcher
    done onComplete  {
      case Success(_) => println("Done"); system.terminate()
      case Failure(err) => println(err.toString); system.terminate()
    }
  }
}

To correctly handle Ctrl+C or if your app is hosted inside a container and you want to gracefully shutdown you can use the following code:

// retrieve the control object
val control = val done = Consumer
  .plainSource(consumerSettings, Subscriptions.topics("test-topic"))
  .toMat(Sink.foreach(println))(Keep.both)
  .mapMaterializedValue(DrainingControl.apply)
  .run()

// Correctly handle Ctrl+C and docker container stop
sys.addShutdownHook({
  println("Shutdown requested...")

  val done = control.shutdown()

  implicit val ec: ExecutionContextExecutor = system.dispatcher
  done
    .onComplete {
      case Success(_) => logger.info("Exiting ...")
      case Failure(err) => logger.error("Error", err)
    }
})

See also

• https://doc.akka.io/docs/akka-stream-kafka/current/home.html
• https://github.com/lightbend/kafka-with-akka-streams-kafka-streams-tutorial