io.warp10:warp10-plugin-kafka
Installation
$ wf g -w /path/to/warp10 io.warp10 warp10-plugin-kafka
/path/to/warp10/
is where Warp 10 is physically located.
Readme
warp10-plugin-kafka
The Warp 10™ Kafka Plugin allows your Warp 10™ instance to consume messages from a Kafka message broker and process those messages using WarpScript™.
Installation
Using WarpFleet™
wf g -d CONFDIR -l LIBDIR io.warp10 warp10-plugin-kafka
Where CONFDIR
is the configuration directory of your Warp 10™ instance (or use -c CONFFILE
with a release prior to 2.1), and LIBDIR
the lib directory where the downloaded jar file should be copied.
From source
git clone https://github.com/senx/warp10-plugin-kafka.git
cd warp10-plugin-kafka
./gradlew -Duberjar shadowJar
Then copy the jar (suffixed with uberjar
) from the build/libs
directory into the lib
directory of your Warp 10™ deployment.
Configuration
The Kafka plugin will scan the directory specified in the kafka.dir
configuration key at the period configured in kafka.period
(expressed in milliseconds, default to one minute), reloading consumer specifications from .mc2
files found in the directory.
Usage
The plugin will interact with Kafka according to specification files. A specification file is a valid WarpScript™ (.mc2
) file producing a map with the following fields:
{
'topics' [ 'topic1' 'topic2' ] // List of Kafka topics to subscribe to
'parallelism' 1 // Number of threads to start for processing the incoming messages. Each thread will handle a certain number of partitions.
'config' { } // Map of Kafka consumer parameters
'macro' <% %> // Macro to execute for each incoming message
'timeout' 10000 // Polling timeout (in ms), if no message is received within this delay, the macro will be called with an empty map as input
}
Once loaded, this specification will trigger the consumption of the specified Kafka topics. You should set the groupid
in the config
map. For each received message, the macro specified in macro
will be called, preserving the execution environment (stack) between calls. The message is pushed onto the stack prior to calling the macro. It is pushed as a map with the following fields:
{
'timestamp' 123 // The record timestamp
'timestampType' 'type' // The type of timestamp, can be one of 'NoTimestampType', 'CreateTime', 'LogAppendTime'
'topic' 'topic_name' // Name of the topic which received the message
'offset' 123 // Offset of the message in 'topic'
'partition' 123 // Id of the partition which received the message
'key' ... // Byte array of the message key
'value' ... // Byte array of the message value
'headers' { } // Map of message headers
}
When the timeout
has expired, instead of a message map like above, the macro is called with an empty map.
The configured macro can perform any WarpScript™ operation. If you wish to store some data, you can use the UPDATE
function.
Path | Size | Creation time |
---|---|---|
KCOMMIT | 1004 bytes | 2021-05-20 |
Versions | Published |
---|---|
1.0.1-uberjar | 2021-05-21 |