Tremor Config to consume json documents from udp and forward them to kafka while applying backpressure in case of error or timeout after > 1s

config.yaml

onramp:
  - id: udp_in
    type: udp
    codec: json
    config:
      host: localhost
      port: 65535
offramp:
  - id: kafka_out
    type: kafka
    codec: json
    config:
      brokers:
        - my_kafka:9092
        - my_other_kafka:9092
      topic: my_topic
binding:
  - id: my_binding
    links:
      "/onramp/udp_in/{instance}/out": ["/pipeline/main/{instance}/in"]
      "/pipeline/main/{instance}/out": ["/offramp/kafka_out/{instance}/in"]
      "/pipeline/main/{instance}/err": ["/offramp/system::stderr/system/in"]
mapping:
  /binding/my_binding/01:
    instance: 01

main.trickle

define qos::backpressure operator bp
with
 timeout = 1000
end;
create operator bp;
 
select event from in into bp;
select event from bp into out;
select event from bp/err into err;