logstash config

aggregate.conf

input {
  file {
    path => "/tmp/test.txt"
  }
}
filter {
    json { source => "message" target => "meta" }
    # To mae sure its not a nested filed issue copy the ExchangeId and Type to an upper level field
    mutate {
       copy => { "[meta][ExchangeId]" => "ExchangeId" }
       copy => { "[meta][Type]" => "Type" }
    }
    aggregate {
        # matching id is ExchangeId
        task_id => "%{ExchangeId}"
        timeout => 15
        push_map_as_event_on_timeout => true
        code => '
          # we want 2 nested objects in the document object REQ_OUT and RESP_IN
          type = event.get("Type")
          map[type] = {}
          event.to_hash.each { |k, v|
            map[type][k] = v
          }
        '
        timeout_task_id_field => "ExchangeId"
        timeout_tags => ['_aggregatetimeout']

    }
    mutate {
        # after the field have been merged we do not need the message and meta block any more (TODO: if there is no agregate exception clause)
        remove_field => [ "meta" , "message" ]
    }
}

output {
  stdout { codec => rubydebug }
}

output

{
          "Type" => "REQ_OUT",
      "@version" => "1",
          "host" => "LGS02",
    "ExchangeId" => "260e06a3-9cb5-4154-bf97-637e929fa4c2",
    "@timestamp" => 2021-04-14T06:46:29.221Z,
          "path" => "/tmp/test.txt"
}
{
          "Type" => "RESP_IN",
      "@version" => "1",
          "host" => "LGS02",
    "ExchangeId" => "260e06a3-9cb5-4154-bf97-637e929fa4c2",
    "@timestamp" => 2021-04-14T06:46:29.247Z,
          "path" => "/tmp/test.txt"
}

test.txt

{"timestamp":"2021-04-02T05:50:44.251Z", "severity":"INFO","Type":"REQ_OUT","Address":"https://a-link.com", "ExchangeId":"260e06a3-9cb5-4154-bf97-637e929fa4c2","REQ_OUT":"test1"}
{"timestamp":"2021-04-02T05:50:45.534Z","severity":"INFO","Type":"RESP_IN","ResponseCode":"200","Address":"https://a-link.com","ExchangeId":"260e06a3-9cb5-4154-bf97-637e929fa4c2","RESP_IN":"test2"}