How to make http processor parameters come from kafka input

[tri0mphe]

input:
  label: ""
  kafka_franz:
    .........
buffer:
  none: {}
pipeline:
  threads: 2
  processors: 
    - mapping: |
        root.src_ip = this.src_ip
    - branch:
        processors:
          - http:
              url: "http://xxx.com/api/v1/ipUserQuery?ip=${! root.src_ip }&needIpType=true&needHistory=true&needUserInfo=true" # No default (required)
              verb: GET
              timeout: 5s
              headers:
                Accept: "application/json"
        result_map: "root.response = this"

the input is kafka, kafka result is in root.src_ip = this.src_ip ，I want use kafka input src_ip to get http reqeust${! root.src_ip }。
but this is can't run。

error is：

ERRO HTTP request to 'http://xxx.com/api/v1/ipUserQuery?ip=${! root.src_ip }&needIpType=true&needHistory=true&needUserInfo=true' failed: url interpolation error: unable to reference `root` from this context  @service=redpanda-connect label="" path=root.pipeline.processors.1.branch.processors.0

[tri0mphe]

I have solved the problem

pipeline:
  threads: 2
  processors: 
    - mapping: |
        meta src_ip = this.src_ip
    - branch:
        request_map: 'root = ""'
        processors:
          - http:
              url: 'http://test.com/api/v1/ipUserQuery?ip=${! meta("src_ip") }&needIpType=true&needHistory=true&needUserInfo=true' # No default (required)
              verb: GET
              timeout: 5s
              headers:
                Accept: "application/json"
        result_map: "root.response = this"