= Vert.x Kafka client
:toc: left
:lang: groovy
:groovy: groovy

This component provides a Kafka client for reading and sending messages from/to an link:https://kafka.apache.org/[Apache Kafka] cluster.

As consumer, the API provides methods for subscribing to a topic partition receiving
messages asynchronously or reading them as a stream (even with the possibility to pause/resume the stream).

As producer, the API provides methods for sending message to a topic partition like writing on a stream.

WARNING: this module has the tech preview status, this means the API can change between versions.

== Using the Vert.x Kafka client

As component not yet officially released in the Vert.x stack, to use the Vert.x Kafka client current snapshot version,
add the following repository under the _repositories_ section and the following dependency to the _dependencies_ section
of your build descriptor:

* Maven (in your `pom.xml`):

[source,xml,subs="+attributes"]
----
<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-kafka-client</artifactId>
  <version>3.4.0</version>
</dependency>
----

* Gradle (in your `build.gradle` file):

[source,groovy,subs="+attributes"]
----
compile io.vertx:vertx-kafka-client:3.4.0
----

== Creating Kafka clients

Creating consumers and sproducer is quite similar and on how it works using the native Kafka client library.

They need to be configured with a bunch of properties as described in the official
Apache Kafka documentation, for the link:https://kafka.apache.org/documentation/#newconsumerconfigs[consumer] and
for the link:https://kafka.apache.org/documentation/#producerconfigs[producer].

To achieve that, a map can be configured with such properties passing it to one of the
static creation methods exposed by `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html[KafkaConsumer]` and
`link:../../apidocs/io/vertx/kafka/client/producer/KafkaProducer.html[KafkaProducer]`

[source,groovy]
----

// creating the consumer using map config
def config = [:]
config["bootstrap.servers"] = "localhost:9092"
config["key.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
config["value.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
config["group.id"] = "my_group"
config["auto.offset.reset"] = "earliest"
config["enable.auto.commit"] = "false"

// use consumer for interacting with Apache Kafka
def consumer = KafkaConsumer.create(vertx, config)

----

In the above example, a `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html[KafkaConsumer]` instance is created using
a map instance in order to specify the Kafka nodes list to connect (just one) and
the deserializers to use for getting key and value from each received message.

Likewise a producer can be created

[source,groovy]
----

// creating the producer using map and class types for key and value serializers/deserializers
def config = [:]
config["bootstrap.servers"] = "localhost:9092"
config["key.serializer"] = "org.apache.kafka.common.serialization.StringSerializer"
config["value.serializer"] = "org.apache.kafka.common.serialization.StringSerializer"
config["acks"] = "1"

// use producer for interacting with Apache Kafka
def producer = KafkaProducer.create(vertx, config)

----

ifdef::java,groovy,kotlin[]
Another way is to use a `link:../../apidocs/java/util/Properties.html[Properties]` instance instead of the map.

[source,groovy]
----

// creating the consumer using properties config
def config = new java.util.Properties()
config.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
config.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class)
config.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class)
config.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, "my_group")
config.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
config.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")

// use consumer for interacting with Apache Kafka
def consumer = KafkaConsumer.create(vertx, config)

----

More advanced creation methods allow to specify the class type for the key and the value used for sending messages
or provided by received messages; this is a way for setting the key and value serializers/deserializers instead of
using the related properties for that

[source,groovy]
----

// creating the producer using map and class types for key and value serializers/deserializers
def config = new java.util.Properties()
config.put(org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
config.put(org.apache.kafka.clients.producer.ProducerConfig.ACKS_CONFIG, "1")

// use producer for interacting with Apache Kafka
def producer = KafkaProducer.create(vertx, config, java.lang.String.class, java.lang.String.class)

----

Here the `link:../../apidocs/io/vertx/kafka/client/producer/KafkaProducer.html[KafkaProducer]` instance is created in using a `link:../../apidocs/java/util/Properties.html[Properties]` for
specifying Kafka nodes list to connect (just one) and the acknowledgment mode; the key and value deserializers are
specified as parameters of `link:../../apidocs/io/vertx/kafka/client/producer/KafkaProducer.html#create-io.vertx.core.Vertx-java.util.Properties-java.lang.Class-java.lang.Class-[KafkaProducer.create]`.
endif::[]

== Receiving messages from a topic joining a consumer group

In order to start receiving messages from Kafka topics, the consumer can use the
`link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#subscribe-java.util.Set-[subscribe]` method for
subscribing to a set of topics being part of a consumer group (specified by the properties on creation).

You need to register an handler for handling incoming messages using the
`link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#handler-io.vertx.core.Handler-[handler]`

[source,groovy]
----

// register the handler for incoming messages
consumer.handler({ record ->
  println("Processing key=${record.key()},value=${record.value()},partition=${record.partition()},offset=${record.offset()}")
})

// subscribe to several topics
def topics = new java.util.HashSet()
topics.add("topic1")
topics.add("topic2")
topics.add("topic3")
consumer.subscribe(topics)

// or just subscribe to a single topic
consumer.subscribe("a-single-topic")

----

An handler can also be passed during subscription to be aware of the subscription result and being notified when the operation
is completed.

[source,groovy]
----

// register the handler for incoming messages
consumer.handler({ record ->
  println("Processing key=${record.key()},value=${record.value()},partition=${record.partition()},offset=${record.offset()}")
})

// subscribe to several topics
def topics = new java.util.HashSet()
topics.add("topic1")
topics.add("topic2")
topics.add("topic3")
consumer.subscribe(topics, { ar ->
  if (ar.succeeded()) {
    println("subscribed")
  } else {
    println("Could not subscribe ${ar.cause().getMessage()}")
  }
})

// or just subscribe to a single topic
consumer.subscribe("a-single-topic", { ar ->
  if (ar.succeeded()) {
    println("subscribed")
  } else {
    println("Could not subscribe ${ar.cause().getMessage()}")
  }
})

----

Using the consumer group way, the Kafka cluster assigns partitions to the consumer taking into account other connected
consumers in the same consumer group, so that partitions can be spread across them.

The Kafka cluster handles partitions re-balancing when a consumer leaves the group (so assigned partitions are free
to be assigned to other consumers) or a new consumer joins the group (so it wants partitions to read from).

You can register handlers on a `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html[KafkaConsumer]` to be notified
of the partitions revocations and assignments by the Kafka cluster using
`link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#partitionsRevokedHandler-io.vertx.core.Handler-[partitionsRevokedHandler]` and
`link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#partitionsAssignedHandler-io.vertx.core.Handler-[partitionsAssignedHandler]`.

[source,groovy]
----

// register the handler for incoming messages
consumer.handler({ record ->
  println("Processing key=${record.key()},value=${record.value()},partition=${record.partition()},offset=${record.offset()}")
})

// registering handlers for assigned and revoked partitions
consumer.partitionsAssignedHandler({ topicPartitions ->

  println("Partitions assigned")
  topicPartitions.each { topicPartition ->
    println("${topicPartition.topic} ${topicPartition.partition}")
  }
})

consumer.partitionsRevokedHandler({ topicPartitions ->

  println("Partitions revoked")
  topicPartitions.each { topicPartition ->
    println("${topicPartition.topic} ${topicPartition.partition}")
  }
})

// subscribes to the topic
consumer.subscribe("test", { ar ->

  if (ar.succeeded()) {
    println("Consumer subscribed")
  }
})

----

After joining a consumer group for receiving messages, a consumer can decide to leave the consumer group in order to
not get messages anymore using `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#unsubscribe--[unsubscribe]`

[source,groovy]
----

// consumer is already member of a consumer group

// unsubscribing request
consumer.unsubscribe()

----

You can add an handler to be notified of the result

[source,groovy]
----

// consumer is already member of a consumer group

// unsubscribing request
consumer.unsubscribe({ ar ->

  if (ar.succeeded()) {
    println("Consumer unsubscribed")
  }
})

----

== Receiving messages from a topic requesting specific partitions

Besides being part of a consumer group for receiving messages from a topic, a consumer can ask for a specific
topic partition. When the consumer is not part part of a consumer group the overall application cannot
rely on the re-balancing feature.

You can use `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#assign-java.util.Set-io.vertx.core.Handler-[assign]`
in order to ask for specific partitions.

[source,groovy]
----

// register the handler for incoming messages
consumer.handler({ record ->
  println("key=${record.key()},value=${record.value()},partition=${record.partition()},offset=${record.offset()}")
})

//
def topicPartitions = new java.util.HashSet()
topicPartitions.add([
  topic:"test",
  partition:0
])

// requesting to be assigned the specific partition
consumer.assign(topicPartitions, { done ->

  if (done.succeeded()) {
    println("Partition assigned")

    // requesting the assigned partitions
    consumer.assignment({ done1 ->

      if (done1.succeeded()) {

        done1.result().each { topicPartition ->
          println("${topicPartition.topic} ${topicPartition.partition}")
        }
      }
    })
  }
})

----

Calling `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#assignment-io.vertx.core.Handler-[assignment]` provides
the list of the current assigned partitions.

== Getting topic partition information

You can call the `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#partitionsFor-java.lang.String-io.vertx.core.Handler-[partitionsFor]` to get information about
partitions for a specified topic

[source,groovy]
----

// asking partitions information about specific topic
consumer.partitionsFor("test", { ar ->

  if (ar.succeeded()) {

    ar.result().each { partitionInfo ->
      println(partitionInfo)
    }
  }
})

----

In addition `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#listTopics-io.vertx.core.Handler-[listTopics]` provides all available topics
with related partitions

[source,groovy]
----

// asking information about available topics and related partitions
consumer.listTopics({ ar ->

  if (ar.succeeded()) {

    def map = ar.result()
    map.each { topic, partitions ->
      println("topic = ${topic}")
      println("partitions = ${map[topic]}")
    }
  }
})

----

== Manual offset commit

In Apache Kafka the consumer is in charge to handle the offset of the last read message.

This is executed by the commit operation executed automatically every time a bunch of messages are read
from a topic partition. The configuration parameter `enable.auto.commit` must be set to `true` when the
consumer is created.

Manual offset commit, can be achieved with `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#commit-io.vertx.core.Handler-[commit]`.
It can be used to achieve _at least once_ delivery to be sure that the read messages are processed before committing
the offset.

[source,groovy]
----

// consumer is processing read messages

// committing offset of the last read message
consumer.commit({ ar ->

  if (ar.succeeded()) {
    println("Last read message offset committed")
  }
})

----

== Seeking in a topic partition

Apache Kafka can retain messages for a long period of time and the consumer can seek inside a topic partition
and obtain arbitraty access to the messages.

You can use `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#seek-io.vertx.kafka.client.common.TopicPartition-long-[seek]` to change the offset for reading at a specific
position

[source,groovy]
----

def topicPartition = [
  topic:"test",
  partition:0
]

// seek to a specific offset
consumer.seek(topicPartition, 10, { done ->

  if (done.succeeded()) {
    println("Seeking done")
  }
})


----

When the consumer needs to re-read the stream from the beginning, it can use `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#seekToBeginning-io.vertx.kafka.client.common.TopicPartition-[seekToBeginning]`

[source,groovy]
----

def topicPartition = [
  topic:"test",
  partition:0
]

// seek to the beginning of the partition
consumer.seekToBeginning(java.util.Collections.singleton(topicPartition), { done ->

  if (done.succeeded()) {
    println("Seeking done")
  }
})

----

Finally `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#seekToEnd-io.vertx.kafka.client.common.TopicPartition-[seekToEnd]` can be used to come back at the end of the partition

[source,groovy]
----

def topicPartition = [
  topic:"test",
  partition:0
]

// seek to the end of the partition
consumer.seekToEnd(java.util.Collections.singleton(topicPartition), { done ->

  if (done.succeeded()) {
    println("Seeking done")
  }
})

----

== Message flow control

A consumer can control the incoming message flow and pause/resume the read operation from a topic, e.g it
can pause the message flow when it needs more time to process the actual messages and then resume
to continue message processing.

To achieve that you can use `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#pause--[pause]` and
`link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#resume--[resume]`

[source,groovy]
----

def topicPartition = [
  topic:"test",
  partition:0
]

// registering the handler for incoming messages
consumer.handler({ record ->
  println("key=${record.key()},value=${record.value()},partition=${record.partition()},offset=${record.offset()}")

  // i.e. pause/resume on partition 0, after reading message up to offset 5
  if ((record.partition() == 0) && (record.offset() == 5)) {

    // pause the read operations
    consumer.pause(topicPartition, { ar ->

      if (ar.succeeded()) {

        println("Paused")

        // resume read operation after a specific time
        vertx.setTimer(5000, { timeId ->

          // resumi read operations
          consumer.resume(topicPartition)
        })
      }
    })
  }
})

----

== Closing a consumer

Call close to close the consumer. Closing the consumer closes any open connections and releases all consumer resources.

The close is actually asynchronous and might not complete until some time after the call has returned. If you want to be notified
when the actual close has completed then you can pass in a handler.

This handler will then be called when the close has fully completed.

[source,groovy]
----
consumer.close({ res ->
  if (res.succeeded()) {
    println("Consumer is now closed")
  } else {
    println("close failed")
  }
})

----

== Sending messages to a topic

You can use  `link:../../apidocs/io/vertx/kafka/client/producer/KafkaProducer.html#write-io.vertx.kafka.client.producer.KafkaProducerRecord-[write]` to send messages (records) to a topic.

The simplest way to send a message is to specify only the destination topic and the related value, omitting its key
or partition, in this case the messages are sent in a round robin fashion across all the partitions of the topic.

[source,groovy]
----

(0..<5).each { i ->

  // only topic and message value are specified, round robin on destination partitions
  def record = KafkaProducerRecord.create("test", "message_${i}")

  producer.write(record)
}

----

You can receive message sent metadata like its topic, its destination partition and its assigned offset.

[source,groovy]
----

(0..<5).each { i ->

  // only topic and message value are specified, round robin on destination partitions
  def record = KafkaProducerRecord.create("test", "message_${i}")

  producer.write(record, { done ->

    if (done.succeeded()) {

      def recordMetadata = done.result()
      println("Message ${record.value()} written on topic=${recordMetadata.topic}, partition=${recordMetadata.partition}, offset=${recordMetadata.offset}")
    }

  })
}


----

When you need to assign a partition to a message, you can specify its partition identifier
or its key

[source,groovy]
----

(0..<10).each { i ->

  // a destination partition is specified
  def record = KafkaProducerRecord.create("test", null, "message_${i}", 0)

  producer.write(record)
}

----

Since the producers identifies the destination using key hashing, you can use that to guarantee that all
messages with the same key are sent to the same partition and retain the order.

[source,groovy]
----

(0..<10).each { i ->

  // i.e. defining different keys for odd and even messages
  def key = i % 2

  // a key is specified, all messages with same key will be sent to the same partition
  def record = KafkaProducerRecord.create("test", java.lang.String.valueOf(key), "message_${i}")

  producer.write(record)
}

----

NOTE: the shared producer is created on the first `createShared` call and its configuration is defined at this moment,
shared producer usage must use the same configuration.

== Sharing a producer

Sometimes you want to share the same producer from within several verticles or contexts.

Calling `link:../../apidocs/io/vertx/kafka/client/producer/KafkaProducer.html#createShared-io.vertx.core.Vertx-java.lang.String-java.util.Map-[KafkaProducer.createShared]`
returns a producer that can be shared safely.

[source,groovy]
----

// Create a shared producer identified by 'the-producer'
def producer1 = KafkaProducer.createShared(vertx, "the-producer", config)

// Sometimes later you can close it
producer1.close()

----

The same resources (thread, connection) will be shared between the producer returned by this method.

When you are done with the producer, just close it, when all shared producers are closed, the resources will
be released for you.

== Closing a producer

Call close to close the producer. Closing the producer closes any open connections and releases all producer resources.

The close is actually asynchronous and might not complete until some time after the call has returned. If you want to be notified
when the actual close has completed then you can pass in a handler.

This handler will then be called when the close has fully completed.

[source,groovy]
----
producer.close({ res ->
  if (res.succeeded()) {
    println("Producer is now closed")
  } else {
    println("close failed")
  }
})

----

== Getting topic partition information

You can call the `link:../../apidocs/io/vertx/kafka/client/producer/KafkaProducer.html#partitionsFor-java.lang.String-io.vertx.core.Handler-[partitionsFor]` to get information about
partitions for a specified topic:

[source,groovy]
----

// asking partitions information about specific topic
producer.partitionsFor("test", { ar ->

  if (ar.succeeded()) {

    ar.result().each { partitionInfo ->
      println(partitionInfo)
    }
  }
})

----

== Handling errors

Errors handling (e.g timeout) between a Kafka client (consumer or producer) and the Kafka cluster is done using
`link:../../apidocs/io/vertx/kafka/client/consumer/KafkaConsumer.html#exceptionHandler-io.vertx.core.Handler-[exceptionHandler]` or
`link:../../apidocs/io/vertx/kafka/client/producer/KafkaProducer.html#exceptionHandler-io.vertx.core.Handler-[exceptionHandler]`

[source,groovy]
----

// setting handler for errors
consumer.exceptionHandler({ e ->
  println("Error = ${e.getMessage()}")
})

----

== Automatic clean-up in verticles

If you’re creating consumers and producer from inside verticles, those consumers and producers will be automatically
closed when the verticle is undeployed.

== Using Vert.x serializers/deserizaliers

Vert.x Kafka client comes out of the box with serializers and deserializers for buffers, json object
and json array.

In a consumer you can use buffers

[source,groovy]
----

// Creating a consumer able to deserialize to buffers
def config = [:]
config["bootstrap.servers"] = "localhost:9092"
config["key.deserializer"] = "io.vertx.kafka.client.serialization.BufferDeserializer"
config["value.deserializer"] = "io.vertx.kafka.client.serialization.BufferDeserializer"
config["group.id"] = "my_group"
config["auto.offset.reset"] = "earliest"
config["enable.auto.commit"] = "false"

// Creating a consumer able to deserialize to json object
config = [:]
config["bootstrap.servers"] = "localhost:9092"
config["key.deserializer"] = "io.vertx.kafka.client.serialization.JsonObjectDeserializer"
config["value.deserializer"] = "io.vertx.kafka.client.serialization.JsonObjectDeserializer"
config["group.id"] = "my_group"
config["auto.offset.reset"] = "earliest"
config["enable.auto.commit"] = "false"

// Creating a consumer able to deserialize to json array
config = [:]
config["bootstrap.servers"] = "localhost:9092"
config["key.deserializer"] = "io.vertx.kafka.client.serialization.JsonArrayDeserializer"
config["value.deserializer"] = "io.vertx.kafka.client.serialization.JsonArrayDeserializer"
config["group.id"] = "my_group"
config["auto.offset.reset"] = "earliest"
config["enable.auto.commit"] = "false"

----

Or in a producer

[source,groovy]
----

// Creating a producer able to serialize to buffers
def config = [:]
config["bootstrap.servers"] = "localhost:9092"
config["key.serializer"] = "io.vertx.kafka.client.serialization.BufferSerializer"
config["value.serializer"] = "io.vertx.kafka.client.serialization.BufferSerializer"
config["acks"] = "1"

// Creating a producer able to serialize to json object
config = [:]
config["bootstrap.servers"] = "localhost:9092"
config["key.serializer"] = "io.vertx.kafka.client.serialization.JsonObjectSerializer"
config["value.serializer"] = "io.vertx.kafka.client.serialization.JsonObjectSerializer"
config["acks"] = "1"

// Creating a producer able to serialize to json array
config = [:]
config["bootstrap.servers"] = "localhost:9092"
config["key.serializer"] = "io.vertx.kafka.client.serialization.JsonArraySerializer"
config["value.serializer"] = "io.vertx.kafka.client.serialization.JsonArraySerializer"
config["acks"] = "1"

----

ifdef::java,groovy,kotlin[]
You can also specify the serizalizers/deserializers at creation time:

In a consumer

[source,groovy]
----
Code not translatable
----

Or in a producer

[source,groovy]
----
Code not translatable
----

endif::[]

ifdef::java[]
== RxJava API

The Kafka client provides an Rxified version of the original API.

[source,groovy]
----
Code not translatable
----
endif::[]

ifdef::java,groovy,kotlin[]
== Stream implementation and native Kafka objects

When you want to operate on native Kafka records you can use a stream oriented
implementation which handles native Kafka objects.

The `link:../../apidocs/io/vertx/kafka/client/consumer/KafkaReadStream.html[KafkaReadStream]` shall be used for reading topic partitions, it is
a read stream of `link:../../apidocs/org/apache/kafka/clients/consumer/ConsumerRecord.html[ConsumerRecord]` objects.

The `link:../../apidocs/io/vertx/kafka/client/producer/KafkaWriteStream.html[KafkaWriteStream]` shall be used for writing to topics, it is a write
stream of `link:../../apidocs/org/apache/kafka/clients/producer/ProducerRecord.html[ProducerRecord]`.

The API exposed by these interfaces is mostly the same than the polyglot version.
endif::[]