Reading Data From Kafka and use filter json fails with ParserError

That is not valid JSON. I would handle that event with dissect and kv. The separators in dissect and the field_split character are tabs.

  dissect { mapping => { "message" => "%{ts}    %{+ts}  %{restOfLine}" } }
  kv { source => "restOfLine"  field_split => " " }