KAFKA::CONSUMER

Consume messages from a Kafka topic.

Prerequisite

KAFKA::CONSUMER Resource

Syntax

actions:
  <action-id>:
    type: KAFKA::CONSUMER
    props:
      topic: <topic_name>
      numOfMessages: <number_of_messages>
      seekToOffset: <offset>
      partition: <partition_number>
      timeout: <timeout_in_milliseconds>

Properties

Property	Description	Type
<action-id>	Unique identifier for the action	string
type	Type of the action	string
props	Properties specific to this action. Refer to props section for more details.	object

props

Property	Description	Type	Required
topic	The Kafka topic from which messages are to be consumed.	string	true
numOfMessages	The number of messages to consume.	number	true
seekToOffset	The offset from which to start consuming messages.	number
partition	The partition from which to consume messages.	number
timeout	Timeout for consuming messages in milliseconds. Default value is 3000 ms.	number

Example

kafkaConsumerTest.yml

actions:
  publish:
    type: KAFKA::CONSUMER
    props:
      topic: example_topic
      value: "Hello, World!"

tests:
  kafkaConsumerTest:
    tasks:
      - id: task1
        resource: resources.kafka
        action: actions.publish

Output

📝 Note: Can be accessed using context["last_output"]

Key	Description	Type
messages	The messages consumed from the Kafka topic.	array
timeTaken	Time taken for the consume operation, in milliseconds.	long

Example

{
  "messages": [
    // ... consumed messages ...
  ],
  "timeTaken": 5002
}