DeduplicateItems Fails on ion input about kestra HOT 3 CLOSED

triggers:
  - id: consumetrigger
    type: io.kestra.plugin.kafka.Trigger
    groupId: kestra-nonprodmap
    properties:
      auto.offset.reset: latest
      bootstrap.servers: "{{ secret('KAFKA_HOST') }}"
      sasl.jaas.config: >-
        org.apache.kafka.common.security.scram.ScramLoginModule required 
        username="{{ secret('KAFKA_USER') }}" 
        password="{{ secret('KAFKA_PASSWORD') }}";
      sasl.mechanism: SCRAM-SHA-512
      security.protocol: SASL_PLAINTEXT
    topicPattern: ^.*event.cnc.map
    keyDeserializer: STRING
    valueDeserializer: JSON
    # Very Important! The default pollDuration of 2.0 seconds is not long enough
    pollDuration: 10.0

tasks:
 # This fails - I can get it to work by passing trigger.uri into a jsonWriter and then reading outputs.json.uri into dedup....
  - id: dedup
    type: io.kestra.core.tasks.storages.DeduplicateItems
    from: "{{ trigger.uri }}"
    expr: "{{ key }}"