This library allow to consume kafka messages and handle them making HTTP REST calls.
To include the library as a dependency of your Maven/Gradle project use the following snippets.
You need to add Fuel repository to your
First of all you need to initialize Kafka2Rest with:
<K, V>used in library ConsumerRecords.
- Kafka Consumer. Remember to set false "enable.auto.commit" property if you need to handle processors failures.
- A Config object.
- Kafka Producer (Optional). This parameter must be set in order to use the
|maxIntervalBetweenPolls||The maximum time (Duration) interval between two consecutive poll. When this timeout is exceeded the method |
|topicList||The list of topic to subscribe to.|
|pollTimeout||The time (Duration) spent waiting in poll if data is not available in the buffer. If 0, returns immediately with any records that are available currently in the buffer, else returns empty. Must not be negative.|
|onExceptionTopic||(Optional) Defines a topic where messages causing unhandled processor exceptions will be stored. Kafka2Rest will enrich the message headers with two properties |
Then you need to set the Filter And Processor couples. Filter parse kafka message and return boolean .
If true the linked processor run its
process() method and do something.
Just start Kafka2Rest. Invonking the
start() method Kafka2Rest will run on new thread and listen kafka topics.
Kafka2rest offers two method to verify service and connection healthiness.
- isRunning() This method verifies the timeout between two consecutive poll. It returns
truewhen the config
maxIntervalBetweenPollstimeout is respected. This ensures that the consumer is polling and the processors aren't getting stuck.
- isConnected() This method verifies that consumer can list broker topics and it must be connected to kafka server to do this. It returns
truewhen topics are more than 0.
You can also send messages to kafka host building ProducerManager and using the