Can not parse valid JSON message in LogStash

[dencowboy]

I'm using ElasticSearch 7.7 and logstash 7.9.2.

This is my configuration:

input {
  kinesis {
    kinesis_stream_name => "xxx-${ENV}"
    application_name => "xxx-${ENV}"
    region => "eu-west-1"
    tags => ["log"]
  }
}

filter {
  if "log" in [tags] {
    json {
        source => "message"
    }
    mutate {
      add_field => {
        "appName" => "xxx"
        "sourceType" => "kinesis"
        "logstashParser" => "logstash_kinesis_xxx_log"
      }
    }
  }
}

output {
  if "log" in [tags] {
    elasticsearch {
      hosts => "https://${ES}:443"
      index => "ki-xxx-log-%{+YYYY.MM.dd}"
      # https://github.com/elastic/logstash/issues/11094
      ilm_enabled => false
    }
  }
}

Now I see data in my Kibana but it is not parsing the JSON message while the message is a valid json (using jsonlint):

The output looks like below so the fields in my message are not parsed

Any idea what I'm doing wrong or how I can debug this further? I can not find the id of the message in my logstash-plain.log file.

Thanks