​Realtime​Trigger

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