Consume

yaml

type: "io.kestra.plugin.kafka.Consume"

Consume messages from one or more Kafka topics.

Examples

yaml

id: "consume"
type: "io.kestra.plugin.kafka.Consume"
topic: test_kestra
properties:
  bootstrap.servers: localhost:9092
serdeProperties:
  schema.registry.url: http://localhost:8085
keyDeserializer: STRING
valueDeserializer: AVRO

Connect to a Kafka cluster with SSL.

yaml

id: "consume"
type: "io.kestra.plugin.kafka.Consume"
properties:
  security.protocol: SSL
  bootstrap.servers: localhost:19092
  ssl.key.password: my-ssl-password
  ssl.keystore.type: PKCS12
  ssl.keystore.location: my-base64-encoded-keystore
  ssl.keystore.password: my-ssl-password
  ssl.truststore.location: my-base64-encoded-truststore
  ssl.truststore.password: my-ssl-password
topic:
- kestra_workerinstance
keyDeserializer: STRING
valueDeserializer: STRING

Properties

`keyDeserializer`

Type: object
Dynamic: ❓
Required: ✔️

`pollDuration`

Type: string
Dynamic: ✔️
Required: ✔️
Default: 5.000000000
Format: duration

How often to poll for a record.

If no records are available, the maximum wait duration to wait for new records.

`properties`

Type: object
SubType: string
Dynamic: ✔️
Required: ✔️

Kafka connection properties.

The bootstrap.servers property is a minimal required configuration to connect to a Kafka topic. This property can reference any valid Consumer Configs or Producer Configs as key-value pairs.

If you want to pass a truststore or a keystore, you must provide a base64 encoded string for ssl.keystore.location and ssl.truststore.location.

`valueDeserializer`

Type: object
Dynamic: ❓
Required: ✔️

`groupId`

Type: string
Dynamic: ✔️
Required: ❌

Kafka consumer group ID.

Using a consumer group, we will fetch only records that haven't been consumed yet.

`maxDuration`

Type: string
Dynamic: ❌
Required: ❌
Format: duration

The maximum duration to wait for new records before stopping the consumption process.

It's a soft limit evaluated every second.

`maxRecords`

Type: integer
Dynamic: ❌
Required: ❌

The maximum number of records to fetch before stopping the consumption process.

It's a soft limit evaluated every second.

`partitions`

Type: array
SubType: integer
Dynamic: ❌
Required: ❌

Topic partitions to consume messages from.

Manually assign a list of partitions to the consumer.

`serdeProperties`

Type: object
SubType: string
Dynamic: ✔️
Required: ❌

Serializer configuration

Configuration that will be passed to serializer or deserializer. The avro.use.logical.type.converters is always passed when you have any values set to true.

`since`

Type: string
Dynamic: ✔️
Required: ❌

Timestamp of a message to start consuming messages from.

By default, we consume all messages from the topics with no consumer group or depending on the configuration of the auto.offset.reset property. However, you can provide an arbitrary start time. This property is ignored if a consumer group is used. It must be a valid ISO 8601 date.