gistfile1.txt

  template:
    beans:
      - name: kafkaHeaderDeserializer
        type: "#class:org.apache.camel.kamelets.utils.serialization.kafka.KafkaHeaderDeserializer"
        property:
          - key: enabled
            value: '{{deserializeHeaders}}'
      - name: myAggregatorStrategy
        type: "#class:org.apache.camel.processor.aggregate.StringAggregationStrategy"
        property:
          - key: delimiter
            value: ' - '
    from:
      uri: "kafka:{{topic}}"
      parameters:
        brokers: "{{bootstrapServers}}"
        autoCommitEnable: "{{autoCommitEnable}}"
        allowManualCommit: "{{allowManualCommit}}"
        pollOnError: "{{pollOnError}}"
        autoOffsetReset: "{{autoOffsetReset}}"
        groupId: "{{?consumerGroup}}"
      steps:
        - aggregate:
            aggregation-strategy: "{{myAggregatorStrategy}}"
            aggregation-repository: "#class:org.apache.camel.processor.aggregate.MemoryAggregationRepository"
            completion-size: 10
            correlation-expression:
              simple: "${headers[kafka.KEY]}"
        - process:
            ref: "{{kafkaHeaderDeserializer}}"
        - choice:
            when:
              - simple: '${exchangeProperty.CamelAggregationCompleteCurrentGroup} != null'
                steps:
                  - log: "Daje"
        - to: "kamelet:sink"