1. kafka

   input {
     kafka {
       bootstrap_servers => "127.0.0.1:9092"
       topics => ["SCHOOL_AQDL"]
       group_id => "logstash-group"
       consumer_threads => 3
       codec => "json"
     }
   }

2. syslog

   input {
     syslog {
       port => 12345
       codec => cef
       syslog_field => "syslog"
       grok_pattern => "<%{POSINT:priority}>%{SYSLOGTIMESTAMP:timestamp} CUSTOM GROK HERE"
     }
   }