RealtimeTrigger

id: kafka_realtime_trigger
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.value }}"

triggers:
  - id: realtime_trigger
    type: io.kestra.plugin.kafka.RealtimeTrigger
    topic: test_kestra
    properties:
      bootstrap.servers: localhost:9092
    serdeProperties:
      schema.registry.url: http://localhost:8085
    keyDeserializer: STRING
    valueDeserializer: AVRO
    groupId: kafkaConsumerGroupId

id: kafka_realtime_trigger
namespace: company.team

tasks:
  - id: insert_into_mongodb
    type: io.kestra.plugin.mongodb.InsertOne
    connection:
      uri: mongodb://mongoadmin:secret@localhost:27017/?authSource=admin
    database: kestra
    collection: products
    document: |
      {
        "product_id": "{{ trigger.value | jq('.product_id') | first }}",
        "product_name": "{{ trigger.value | jq('.product_name') | first }}",
        "category": "{{ trigger.value | jq('.product_category') | first }}",
        "brand": "{{ trigger.value | jq('.brand') | first }}"
      }

triggers:
  - id: realtime_trigger
    type: io.kestra.plugin.kafka.RealtimeTrigger
    topic: products
    properties:
      bootstrap.servers: localhost:9092
    serdeProperties:
      valueDeserializer: JSON
    groupId: kestraConsumer