___                 _____ ____  ____  ____
  / _ \ _ __   ___ _ _|_   _/ ___||  _ \| __ )
 | | | | '_ \ / _ \ '_ \| | \___ \| | | |  _ \
 | |_| | |_) |  __/ | | | |  ___) | |_| | |_) |
  \___/| .__/ \___|_| |_|_| |____/|____/|____/
       |_|    The modern time series database.

This plugin allows OpenTSDB to consume messages from a Kafka cluster and write them directly to storage, bypassing the Telnet style or HTTP APIs. It includes a Storage Exception Handler plugin that will post messages back to a Kafka queue if writing to storage fails.

1. Download the source code and run mvn package to create the jar in the target/ directory. Copy this file to your OpenTSDB plugin directory as defined in the opentsdb config via tsd.core.plugin_path.
2. Setup the appropriate Kafka topics and partitions. More on that later.
3. Add configuration settings to your opentsdb.conf file as described later on.
4. Restart the TSD and make sure the plugin was loaded and associated with the proper ID. E.g. look in the logs for lines like:

2017-07-10 23:08:58,264 INFO  [main] RpcManager: Successfully initialized plugin [net.opentsdb.tsd.KafkaRpcPlugin] version: 2.4.0
2017-07-10 23:08:57,790 INFO  [main] TSDB: Successfully initialized storage exception handler plugin [net.opentsdb.tsd.KafkaStorageExceptionHandler] version: 2.4.0

The plugin can accept various message formats using classes implementing the net.opentsdb.data.deserializers.Deserializer interface. A Deserializer must be defined per consumer group and all messages in the topics must be of the same format.

For default JSON formatted messages, use the net.opentsdb.data.deserializers.JSONDeserializer. Each JSON message must include a type field with the value being one of the following:

• Metric A single numeric measurement.
• Aggregate A single numeric measurement that may be a pre-aggregate, a rolled up data point or both.
• Histogram A single histogram measurement.

Each message is similar to the HTTP JSON messages in the OpenTSDB API with the addition of the type field so that the JSON deserializer can figure out what the message contains.

The deserialization class must return a list of typed objects as defined below (with JSON provided as an example).

###Metric

The metric message appears as follows:

{
	"type": "Metric",
	"metric": "sys.cpu.user",
	"tags": {
		"host": "web01"
	},
	"timestamp": 1492641000,
	"value": 42
}

For field information, see [/api/put] (http://opentsdb.net/docs/build/html/api_http/put.html).

###Aggregate

Aggregate messages are the same as those documented in [/api/rollup] (http://opentsdb.net/docs/build/html/api_http/rollup.html).

###Histogram

Histogram messages are documented at [/api/histogram] (http://opentsdb.net/docs/build/html/api_http/histogram.html).

The following properties can be stored in the opentsdb.conf file:

Note the KafkaRpcPlugin.groups and <GROUP_NAME> entries above. Kafka consumers belong to a particular group. The Kafka RPC plugin can launch multiple groups consuming from multiple topics so that OpenTSDB messages can be organized by type or source for more efficient control over rate limits and priorities. When setting the KafkaRpcPlugin.groups value, make sure you have a complete set of KafkaRpcPlugin.<GROUP_NAME>.* parameters per group or initialization will fail.