# Vert.x Kafka client

This component provides a Kafka client for reading and sending messages from/to an 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

To use this component, add the following dependency to the dependencies section of your build descriptor:

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

    <dependency>
     <groupId>io.vertx</groupId>
     <artifactId>vertx-kafka-client</artifactId>
     <version>3.6.3</version>
    </dependency>

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

    compile io.vertx:vertx-kafka-client:3.6.3

## Creating Kafka clients

Creating consumers and producers 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 consumer and for the producer.

To achieve that, a map can be configured with such properties passing it to one of the static creation methods exposed by `KafkaConsumer` and `KafkaProducer`

    require 'vertx-kafka-client/kafka_consumer'

    # creating the consumer using map config
    config = Hash.new()
    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
    consumer = VertxKafkaClient::KafkaConsumer.create(vertx, config)

In the above example, a `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

    require 'vertx-kafka-client/kafka_producer'

    # creating the producer using map and class types for key and value serializers/deserializers
    config = Hash.new()
    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
    producer = VertxKafkaClient::KafkaProducer.create(vertx, config)

## Receiving messages from a topic joining a consumer group

In order to start receiving messages from Kafka topics, the consumer can use the `subscribe` method for subscribing to a set of topics being part of a consumer group (specified by the properties on creation).

You also need to register a handler for handling incoming messages using the `handler`.

    # register the handler for incoming messages
    consumer.handler() { |record|
      puts "Processing key=#{record.key()},value=#{record.value()},partition=#{record.partition()},offset=#{record.offset()}"
    }

    # subscribe to several topics
    topics = Java::JavaUtil::HashSet.new()
    topics.add?("topic1")
    topics.add?("topic2")
    topics.add?("topic3")
    consumer.subscribe(topics)

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

The handler can be registered before or after the call to `subscribe()`; messages won’t be consumed until both methods have been called. This allows you to call `subscribe()`, then `seek()` and finally `handler()` in order to only consume messages starting from a particular offset, for example.

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

    # register the handler for incoming messages
    consumer.handler() { |record|
      puts "Processing key=#{record.key()},value=#{record.value()},partition=#{record.partition()},offset=#{record.offset()}"
    }

    # subscribe to several topics
    topics = Java::JavaUtil::HashSet.new()
    topics.add?("topic1")
    topics.add?("topic2")
    topics.add?("topic3")
    consumer.subscribe(topics) { |ar_err,ar|
      if (ar_err == nil)
        puts "subscribed"
      else
        puts "Could not subscribe #{ar_err.get_message()}"
      end
    }

    # or just subscribe to a single topic
    consumer.subscribe("a-single-topic") { |ar_err,ar|
      if (ar_err == nil)
        puts "subscribed"
      else
        puts "Could not subscribe #{ar_err.get_message()}"
      end
    }

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 `KafkaConsumer` to be notified of the partitions revocations and assignments by the Kafka cluster using `partitionsRevokedHandler` and `partitionsAssignedHandler`.

    # register the handler for incoming messages
    consumer.handler() { |record|
      puts "Processing key=#{record.key()},value=#{record.value()},partition=#{record.partition()},offset=#{record.offset()}"
    }

    # registering handlers for assigned and revoked partitions
    consumer.partitions_assigned_handler() { |topicPartitions|

      puts "Partitions assigned"
      topicPartitions.each do |topicPartition|
        puts "#{topicPartition['topic']} #{topicPartition['partition']}"
      end
    }

    consumer.partitions_revoked_handler() { |topicPartitions|

      puts "Partitions revoked"
      topicPartitions.each do |topicPartition|
        puts "#{topicPartition['topic']} #{topicPartition['partition']}"
      end
    }

    # subscribes to the topic
    consumer.subscribe("test") { |ar_err,ar|

      if (ar_err == nil)
        puts "Consumer subscribed"
      end
    }

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 `unsubscribe`

    # consumer is already member of a consumer group

    # unsubscribing request
    consumer.unsubscribe()

You can add an handler to be notified of the result

    # consumer is already member of a consumer group

    # unsubscribing request
    consumer.unsubscribe() { |ar_err,ar|

      if (ar_err == nil)
        puts "Consumer unsubscribed"
      end
    }

## 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 `assign` in order to ask for specific partitions.

    # register the handler for incoming messages
    consumer.handler() { |record|
      puts "key=#{record.key()},value=#{record.value()},partition=#{record.partition()},offset=#{record.offset()}"
    }

    #
    topicPartitions = Java::JavaUtil::HashSet.new()
    topicPartitions.add?({
      'topic' => "test",
      'partition' => 0
    })

    # requesting to be assigned the specific partition
    consumer.assign(topicPartitions) { |done_err,done|

      if (done_err == nil)
        puts "Partition assigned"

        # requesting the assigned partitions
        consumer.assignment() { |done1_err,done1|

          if (done1_err == nil)

            done1.each do |topicPartition|
              puts "#{topicPartition['topic']} #{topicPartition['partition']}"
            end
          end
        }
      end
    }

As with `subscribe()`, the handler can be registered before or after the call to `assign()`; messages won’t be consumed until both methods have been called. This allows you to call `assign()`, then `seek()` and finally `handler()` in order to only consume messages starting from a particular offset, for example.

Calling `assignment` provides the list of the current assigned partitions.

## Receiving messages with explicit polling

Other than using the internal polling mechanism in order to receive messages from Kafka, the client can subscribe to a topic, avoiding to register the handler for getting the messages and then using the `poll` method.

In this way, the user application is in charge to execute the poll for getting messages when it needs, for example after processing the previous ones.

    # subscribes to the topic
    consumer.subscribe("test") { |ar_err,ar|

      if (ar_err == nil)
        puts "Consumer subscribed"

        vertx.set_periodic(1000) { |timerId|

          consumer.poll(100) { |ar1_err,ar1|

            if (ar1_err == nil)

              records = ar1
              (0...records.size()).each do |i|
                record = records.record_at(i)
                puts "key=#{record.key()},value=#{record.value()},partition=#{record.partition()},offset=#{record.offset()}"
              end
            end
          }

        }
      end
    }

After subscribing successfully, the application start a periodic timer in order to execute the poll and getting messages from Kafka periodically.

## Changing the subscription or assignment

You can change the subscribed topics, or assigned partitions after you have started to consume messages, simply by calling `subscribe()` or `assign()` again.

Note that due to internal buffering of messages it is possible that the record handler will continue to observe messages from the old subscription or assignment after the `subscribe()` or `assign()` method’s completion handler has been called. This is not the case for messages observed by the batch handler: Once the completion handler has been called it will only observe messages read from the subscription or assignment.

## Getting topic partition information

You can call the `partitionsFor` to get information about partitions for a specified topic

    # asking partitions information about specific topic
    consumer.partitions_for("test") { |ar_err,ar|

      if (ar_err == nil)

        ar.each do |partitionInfo|
          puts partitionInfo
        end
      end
    }

In addition `listTopics` provides all available topics with related partitions

    # asking information about available topics and related partitions
    consumer.list_topics() { |ar_err,ar|

      if (ar_err == nil)

        map = ar
        map.each_pair { |topic,partitions|
          puts "topic = #{topic}"
          puts "partitions = #{map[topic]}"
        }
      end
    }

## 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 `commit`. It can be used to achieve at least once delivery to be sure that the read messages are processed before committing the offset.

    # consumer is processing read messages

    # committing offset of the last read message
    consumer.commit() { |ar_err,ar|

      if (ar_err == nil)
        puts "Last read message offset committed"
      end
    }

## 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 arbitrary access to the messages.

You can use `seek` to change the offset for reading at a specific position

    topicPartition = {
      'topic' => "test",
      'partition' => 0
    }

    # seek to a specific offset
    consumer.seek(topicPartition, 10) { |done_err,done|

      if (done_err == nil)
        puts "Seeking done"
      end
    }

When the consumer needs to re-read the stream from the beginning, it can use `seekToBeginning`

    topicPartition = {
      'topic' => "test",
      'partition' => 0
    }

    # seek to the beginning of the partition
    consumer.seek_to_beginning(Java::JavaUtil::Collections.singleton(topicPartition)) { |done_err,done|

      if (done_err == nil)
        puts "Seeking done"
      end
    }

Finally `seekToEnd` can be used to come back at the end of the partition

    topicPartition = {
      'topic' => "test",
      'partition' => 0
    }

    # seek to the end of the partition
    consumer.seek_to_end(Java::JavaUtil::Collections.singleton(topicPartition)) { |done_err,done|

      if (done_err == nil)
        puts "Seeking done"
      end
    }

Note that due to internal buffering of messages it is possible that the record handler will continue to observe messages read from the original offset for a time after the `seek*()` method’s completion handler has been called. This is not the case for messages observed by the batch handler: Once the `seek*()` completion handler has been called it will only observe messages read from the new offset.

## Offset lookup

You can use the beginningOffsets API introduced in Kafka 0.10.1.1 to get the first offset for a given partition. In contrast to `seekToBeginning`, it does not change the consumer’s offset.

    topicPartitions = Java::JavaUtil::HashSet.new()
    topicPartition = {
      'topic' => "test",
      'partition' => 0
    }
    topicPartitions.add?(topicPartition)

    consumer.beginning_offsets(topicPartitions) { |done_err,done|
      if (done_err == nil)
        results = done
        results.each_pair { |topic,beginningOffset|
          puts "Beginning offset for topic=#{topic['topic']}, partition=#{topic['partition']}, beginningOffset=#{beginningOffset}"
        }
      end
    }

    # Convenience method for single-partition lookup
    consumer.beginning_offsets(topicPartition) { |done_err,done|
      if (done_err == nil)
        beginningOffset = done
        puts "Beginning offset for topic=#{topicPartition['topic']}, partition=#{topicPartition['partition']}, beginningOffset=#{beginningOffset}"
      end
    }

You can use the endOffsets API introduced in Kafka 0.10.1.1 to get the last offset for a given partition. In contrast to `seekToEnd`, it does not change the consumer’s offset.

    topicPartitions = Java::JavaUtil::HashSet.new()
    topicPartition = {
      'topic' => "test",
      'partition' => 0
    }
    topicPartitions.add?(topicPartition)

    consumer.end_offsets(topicPartitions) { |done_err,done|
      if (done_err == nil)
        results = done
        results.each_pair { |topic,endOffset|
          puts "End offset for topic=#{topic['topic']}, partition=#{topic['partition']}, endOffset=#{endOffset}"
        }
      end
    }

    # Convenience method for single-partition lookup
    consumer.end_offsets(topicPartition) { |done_err,done|
      if (done_err == nil)
        endOffset = done
        puts "End offset for topic=#{topicPartition['topic']}, partition=#{topicPartition['partition']}, endOffset=#{endOffset}"
      end
    }

You can use the offsetsForTimes API introduced in Kafka 0.10.1.1 to look up an offset by timestamp, i.e. search parameter is an epoch timestamp and the call returns the lowest offset with ingestion timestamp >= given timestamp.

    Code not translatable

## 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 `pause` and `resume`.

In the case of the partition-specific pause and resume it is possible that the record handler will continue to observe messages from a paused partition for a time after the `pause()` method’s completion handler has been called. This is not the case for messages observed by the batch handler: Once the `pause()` completion handler has been called it will only observe messages from those partitions which are not paused.

    topicPartition = {
      'topic' => "test",
      'partition' => 0
    }

    # registering the handler for incoming messages
    consumer.handler() { |record|
      puts "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_err,ar|

          if (ar_err == nil)

            puts "Paused"

            # resume read operation after a specific time
            vertx.set_timer(5000) { |timeId|

              # resume read operations
              consumer.resume(topicPartition)
            }
          end
        }
      end
    }

## 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.

    consumer.close() { |res_err,res|
      if (res_err == nil)
        puts "Consumer is now closed"
      else
        puts "close failed"
      end
    }

## Sending messages to a topic

You can use `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.

    require 'vertx-kafka-client/kafka_producer_record'

    (0...5).each do |i|

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

      producer.write(record)
    end

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

    require 'vertx-kafka-client/kafka_producer_record'

    (0...5).each do |i|

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

      producer.write(record) { |done_err,done|

        if (done_err == nil)

          recordMetadata = done
          puts "Message #{record.value()} written on topic=#{recordMetadata['topic']}, partition=#{recordMetadata['partition']}, offset=#{recordMetadata['offset']}"
        end

      }
    end

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

    require 'vertx-kafka-client/kafka_producer_record'

    (0...10).each do |i|

      # a destination partition is specified
      record = VertxKafkaClient::KafkaProducerRecord.create("test", nil, "message_#{i}", 0)

      producer.write(record)
    end

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.

    require 'vertx-kafka-client/kafka_producer_record'

    (0...10).each do |i|

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

      # a key is specified, all messages with same key will be sent to the same partition
      record = VertxKafkaClient::KafkaProducerRecord.create("test", Java::JavaLang::String.value_of(key), "message_#{i}")

      producer.write(record)
    end

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 `KafkaProducer.createShared` returns a producer that can be shared safely.

    require 'vertx-kafka-client/kafka_producer'

    # Create a shared producer identified by 'the-producer'
    producer1 = VertxKafkaClient::KafkaProducer.create_shared(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.

    producer.close() { |res_err,res|
      if (res_err == nil)
        puts "Producer is now closed"
      else
        puts "close failed"
      end
    }

## Getting topic partition information

You can call the `partitionsFor` to get information about partitions for a specified topic:

    # asking partitions information about specific topic
    producer.partitions_for("test") { |ar_err,ar|

      if (ar_err == nil)

        ar.each do |partitionInfo|
          puts partitionInfo
        end
      end
    }

## Handling errors

Errors handling (e.g timeout) between a Kafka client (consumer or producer) and the Kafka cluster is done using `exceptionHandler` or `exceptionHandler`

    # setting handler for errors
    consumer.exception_handler() { |e|
      puts "Error = #{e.get_message()}"
    }

## 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/deserializers

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

    # Creating a consumer able to deserialize to buffers
    config = Hash.new()
    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 = Hash.new()
    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 = Hash.new()
    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

    # Creating a producer able to serialize to buffers
    config = Hash.new()
    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 = Hash.new()
    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 = Hash.new()
    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"

# Vert.x Kafka AdminUtils

This component provides a vert.x wrapper around the most important functions of Kafka’s AdminUtils. AdminUtils are used to create, modify, and delete topics. Other functionality covered by AdminUtils, but not this wrapper, includes Partition Management, Broker Configuration management, etc.

## Using the AdminUtils

### Create a topic

You can call `createTopic` to create a topic. Parameters are: topic name, number of partitions, number of replicas, and the usual callback to handle the result. It might return an error, e.g. if the number of requested replicas is greater than the number of brokers.

    require 'vertx/vertx'
    require 'vertx-kafka-client/admin_utils'
    adminUtils = VertxKafkaClient::AdminUtils.create(Vertx::Vertx.vertx(), "localhost:2181", true)
    # Create topic 'myNewTopic' with 2 partition and 1 replicas
    adminUtils.create_topic("myNewTopic", 2, 1) { |result_err,result|
      if (result_err == nil)
        puts "Creation of topic myNewTopic successful!"else
        puts "Creation of topic myNewTopic failed: #{result_err.get_localized_message()}"end
    }

### Delete a topic

You can call `deleteTopic` to delete a topic. Parameters are: topic name, and the usual callback to handle the result. It might return an error, e.g. if the topic does not exist.

    require 'vertx/vertx'
    require 'vertx-kafka-client/admin_utils'
    adminUtils = VertxKafkaClient::AdminUtils.create(Vertx::Vertx.vertx(), "localhost:2181", true)
    # Delete topic 'myNewTopic'
    adminUtils.delete_topic("myNewTopic") { |result_err,result|
      if (result_err == nil)
        puts "Deletion of topic myNewTopic successful!"else
        puts "Deletion of topic myNewTopic failed: #{result_err.get_localized_message()}"end
    }

### Change a topic’s configuration

If you need to update the configuration of a topic, e.g., you want to update the retention policy, you can call `changeTopicConfig` to update a topic. Parameters are: topic name, a Map (String → String) with parameters to be changed, and the usual callback to handle the result. It might return an error, e.g. if the topic does not exist.

    require 'vertx/vertx'
    require 'vertx-kafka-client/admin_utils'
    adminUtils = VertxKafkaClient::AdminUtils.create(Vertx::Vertx.vertx(), "localhost:2181", true)
    # Set retention to 1000 ms and max size of the topic partition to 1 kiByte
    properties = Hash.new()
    properties["delete.retention.ms"] = "1000"
    properties["retention.bytes"] = "1024"
    adminUtils.change_topic_config("myNewTopic", properties) { |result_err,result|
      if (result_err == nil)
        puts "Configuration change of topic myNewTopic successful!"else
        puts "Configuration change of topic myNewTopic failed: #{result_err.get_localized_message()}"end
    }
    }

### Check if a topic exists

If you want to check if a topic exists, you can call `topicExists`. Parameters are: topic name, and the usual callback to handle the result. It might return an error, e.g. if the topic does not exist.

    require 'vertx/vertx'
    require 'vertx-kafka-client/admin_utils'
    adminUtils = VertxKafkaClient::AdminUtils.create(Vertx::Vertx.vertx(), "localhost:2181", true)
    adminUtils.topic_exists("myNewTopic") { |result_err,result|
      if (result_err == nil)
        puts "Topic myNewTopic exists: #{result}"
      else
        puts "Failed to check if topic myNewTopic exists: #{result_err.get_localized_message()}"end
    }