# Ship Kafka logs **URL:** \ **Category:** Logstash\ **Created:** [September 23, 2019, 2:27pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706 "2019-09-23T14:27:49Z")\ **Posts on this page:** 20\ **Page:** 1
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 23, 2019, 2:27pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/1 "2019-09-23T14:27:50Z")
Hello, can anyone please give me some info or point me to the doc on how to ship kafka logs to elk. I did check some which did not work out. I also read that it is slightly different than other server logs that we ship Thanks! ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 24, 2019, 1:41pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/2 "2019-09-24T13:41:27Z")
I have been trying to use below logstash config but I get the error " [2019-09-24T13:39:47,479][ERROR][logstash.agent] Cannot create pipeline {:reason=\>"Expected one of #, -, ", ', } at line 12, column 23 (byte 172) after filter{\n if ([fields][application] == "KafkaDevInt") {\n grok {\n match =\> {"}" Logstash config: ```auto input { beats { port => 5044 } beats { port => 5045 } } filter{ if ([fields][application] == "KafkaDevInt") { grok { match => {%{SYSLOG5424SD} %{LOGLEVEL} %{GREEDYDATA:message}} } } } output { elasticsearch { hosts => "localhost:9200" } } ``` kafka logs are in below format: [2019-09-22 13:38:04,227] TRACE [Controller id=0 epoch=1] Received response {error\_code=0,partitions=[{topic=treds,partition=0,error\_code=0}]} for a request sent to broker IP:PORT (id: 0 rack: null) (state.change.logger) Can someone help me here please. Thanks! ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [September 24, 2019, 2:03pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/3 "2019-09-24T14:03:43Z")
> [@Eva](#): > > grok { match =\> {%{SYSLOG5424SD} %{LOGLEVEL} %{GREEDYDATA:message}} } That should be more like ``` grok { match => { "message" => "pattern" } } ``` ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 25, 2019, 5:27am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/4 "2019-09-25T05:27:12Z")
Thanks, will try that out and update here. ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 26, 2019, 9:51am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/6 "2019-09-26T09:51:20Z")
I'm getting "The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information." error . Below are the logs. I have also referred to [https://www.elastic.co/guide/en/logstash/5.6/shutdown.html](https://www.elastic.co/guide/en/logstash/5.6/shutdown.html) using the flag but that did not help. Please help! ```auto [2019-09-26T09:42:15,565][WARN][logstash.shutdownwatcher] {"inflight_count"=>0, "stalling_thread_info"=>{"other"=>[{"thread_id"=>26, "name"=>"[main]"[...]/vendor/bundle/jruby/1.9/gems/logstash-input-beats-3.1.30-java/lib/logstash/inputs/beats.rb:209:in `run'"}, {"thread_id"=>27, "name"=>"[main]"[...]/vendor/bundle/jruby/1.9/gems/logstash-input-beats-3.1.30-java/lib/logstash/inputs/beats.rb:209:in `run'"}, {"thread_id"=>21, "name"=>"[main]>worker0", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in `lock'"}, {"thread_id"=>22, "name"=>"[main]>worker1", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in `lock'"}, {"thread_id"=>23, "name"=>"[main]>worker2", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in `lock'"}, {"thread_id"=>24, "name"=>"[main]>worker3", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in `lock'"}]}} [2019-09-26T09:42:15,567][ERROR][logstash.shutdownwatcher] The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information. [2019-09-26T09:42:20,546][WARN][logstash.shutdownwatcher] {"inflight_count"=>0, "stalling_thread_info"=>{"other"=>[{"thread_id"=>27, "name"=>"[main]"[...]/vendor/bundle/jruby/1.9/gems/logstash-input-beats-3.1.30-java/lib/logstash/inputs/beats.rb:209:in `run'"}, {"thread_id"=>21, "name"=>"[main]>worker0", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:138:in `lock'"}, {"thread_id"=>22, "name"=>"[main]>worker1", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:138:in `lock'"}, {"thread_id"=>23, "name"=>"[main]>worker2", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in `lock'"}, {"thread_id"=>24, "name"=>"[main]>worker3", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:138:in `lock'"}]}} ``` [2019-09-26T09:42:15,565][WARN][logstash.shutdownwatcher] {"inflight\_count"=\>0, "stalling\_thread\_info"=\>{"other"=\>[{"thread\_id"=\>26, "name"=\>"[main]\"[...]/vendor/bundle/jruby/1.9/gems/logstash-input-beats-3.1.30-java/lib/logstash/inputs/beats.rb:209:in `run'"}, {"thread_id"=>27, "name"=>"[main]"[...]/vendor/bundle/jruby/1.9/gems/logstash-input-beats-3.1.30-java/lib/logstash/inputs/beats.rb:209:in`run'"}, {"thread\_id"=\>21, "name"=\>"[main]\>worker0", "current\_call"=\>"[...]/logstash-core/lib/logstash/util/wrapped\_synchronous\_queue.rb:112:in `lock'"}, {"thread_id"=>22, "name"=>"[main]>worker1", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in`lock'"}, {"thread\_id"=\>23, "name"=\>"[main]\>worker2", "current\_call"=\>"[...]/logstash-core/lib/logstash/util/wrapped\_synchronous\_queue.rb:112:in `lock'"}, {"thread_id"=>24, "name"=>"[main]>worker3", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in`lock'"}]}} [2019-09-26T09:42:15,567][ERROR][logstash.shutdownwatcher] The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information. [2019-09-26T09:42:20,546][WARN][logstash.shutdownwatcher] {"inflight\_count"=\>0, "stalling\_thread\_info"=\>{"other"=\>[{"thread\_id"=\>27, "name"=\>"[main]\"[...]/vendor/bundle/jruby/1.9/gems/logstash-input-beats-3.1.30-java/lib/logstash/inputs/beats.rb:209:in `run'"}, {"thread_id"=>21, "name"=>"[main]>worker0", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:138:in`lock'"}, {"thread\_id"=\>22, "name"=\>"[main]\>worker1", "current\_call"=\>"[...]/logstash-core/lib/logstash/util/wrapped\_synchronous\_queue.rb:138:in `lock'"}, {"thread_id"=>23, "name"=>"[main]>worker2", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in`lock'"}, {"thread\_id"=\>24, "name"=\>"[main]\>worker3", "current\_call"=\>"[...]/logstash-core/lib/logstash/util/wrapped\_synchronous\_queue.rb:138:in `lock'"}]}} ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 26, 2019, 11:34am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/7 "2019-09-26T11:34:27Z")
hello, I'm able to ship the data after some troubleshooting but I see that the logs are not appearing in the timestamp they occurred. Example, in the screenshot below, if you notice the 'message' section the event occurred on 24th September but the "@timestamp" doesn't show 24th September. Can you tell me how to fix this please ![image](https://us1.discourse-cdn.com/elastic/original/3X/f/4/f413ee0f18337db1ee864688c2f276f626afe46d.png) ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 26, 2019, 11:38am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/8 "2019-09-26T11:38:30Z")
To add to, I'm using below grok pattern: grok { match =\> {"actual\_message" =\> "%{SYSLOG5424SD} %{LOGLEVEL} %{GREEDYDATA}"} } Log format: [2019-09-24 12:54:11,546] DEBUG [Controller id=0] Topics not in preferred replica Map() (kafka.controller.KafkaController) [2019-09-24 12:54:11,546] TRACE [Controller id=0] Leader imbalance ratio for broker 0 is 0.0 (kafka.controller.KafkaController) ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [September 26, 2019, 3:38pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/9 "2019-09-26T15:38:11Z")
That grok doesn't add any fields to the message. You need to name them. Then use a date filter to parse the timestamp. ``` grok { match => {"actual_message" => "%{SYSLOG5424SD:[@metadata][date]} %{LOGLEVEL} %{GREEDYDATA}"} } date { match => ["[@metadata][date]", "[YYYY-MM-dd HH:mm:ss,SSS]" ] } ``` ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 27, 2019, 7:43am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/10 "2019-09-27T07:43:24Z")
Thanks! How do i get the '@timestamp' to be in GMT. In the 'message' section the log is showing in GMT but '@timestamp' is pulling standard time of the machine. Also I'm trying to ship all the old logs as well but I see all the old logs in today's date again. I do not understand what is causing this. ![image](https://us1.discourse-cdn.com/elastic/original/3X/6/7/670f6b62577d41a2508fdcbefc9971f0b6100641.png) ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [September 27, 2019, 1:31pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/11 "2019-09-27T13:31:22Z")
Use the [timezone](https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html#plugins-filters-date-timezone) option on the date filter. ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 28, 2019, 8:43am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/12 "2019-09-28T08:43:17Z")
I tried using timezone =\> "UTC" but i still see the same : ![image](https://us1.discourse-cdn.com/elastic/original/3X/6/a/6a773f6094738a020d079f142ff0f9bf8ddb942e.png) Below is my logstash config: ```auto input { beats { port => 5044 } } filter{ if ([fields][application] == "KafkaDevInt") { grok { match => {"actual_message" => "%{SYSLOG5424SD:[@metadata][date]} %{LOGLEVEL} %{GREEDYDATA}"} } date { match => ["[@metadata][date]","[YYYY-MM-dd HH:mm:ss,SSS]" ] timezone => "UTC" } } } output{ elasticsearch { hosts => "localhost:9200" } } ``` ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 28, 2019, 10:02am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/13 "2019-09-28T10:02:15Z")
This is really wierd, with the same config I sometimes see no logs shipping at all and with no errors seen anywhere. Config test also seems to be reporting fine. Please help me sort this out ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [September 28, 2019, 12:37pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/14 "2019-09-28T12:37:48Z")
The timezone option tells the date filter what timezone the time in the log entry is in, not what timezone you want it converted to. It always converts to UTC. ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 28, 2019, 5:04pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/15 "2019-09-28T17:04:28Z")
But the timezone is always UTC in the logs. Wondering why is it not displayed as UTC in @timestamp? ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [September 28, 2019, 5:41pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/16 "2019-09-28T17:41:48Z")
In Kibana, can you click on event then copy and paste all the fields from the JSON tab? ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [September 28, 2019, 6:17pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/18 "2019-09-28T18:17:04Z")
Please do not post pictures of text, just post the text. ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 28, 2019, 6:22pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/19 "2019-09-28T18:22:03Z")
Ok, I removed the image. Kindly check the log lines below : ```auto [2019-09-28T09:35:10,665][INFO][logstash.inputs.beats] Beats inputs: Starting input listener {:address=>"0.0.0.0:5044"} [2019-09-28T09:35:10,694][INFO][logstash.pipeline] Pipeline main started [2019-09-28T09:35:10,702][INFO][org.logstash.beats.Server] Starting server on port: 5044 [2019-09-28T09:35:10,735][INFO][logstash.agent] Successfully started Logstash API endpoint {:port=>9600} [2019-09-28T10:31:11,263][WARN][logstash.runner] SIGTERM received. Shutting down the agent. [2019-09-28T10:31:11,267][WARN][logstash.agent] stopping pipeline {:id=>"main"} [2019-09-28T10:31:16,301][WARN][logstash.shutdownwatcher] {"inflight_count"=>0, "stalling_thread_info"=>{"other"=>[{"thread_id"=>26, "name"=>"[main]"[...]/vendor/bundle/jruby/1.9/gems/logstash-input-beats-3.1.32-java/lib/logstash/inputs/beats.rb:209:in `run'"}, {"thread_id"=>21, "name"=>"[main]>worker0", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:138:in `lock'"}, {"thread_id"=>22, "name"=>"[main]>worker1", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:112:in `lock'"}, {"thread_id"=>23, "name"=>"[main]>worker2", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:138:in `lock'"}, {"thread_id"=>24, "name"=>"[main]>worker3", "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:138:in `lock'"}]}} [2019-09-28T10:31:16,302][ERROR][logstash.shutdownwatcher] The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information. [2019-09-28T10:31:24,890][INFO][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/usr/share/logstash/modules/fb_apache/configuration"} [2019-09-28T10:31:24,893][INFO][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/usr/share/logstash/modules/netflow/configuration"} [2019-09-28T10:31:25,402][INFO][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}} [2019-09-28T10:31:25,404][INFO][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://localhost:9200/, :path=>"/"} [2019-09-28T10:31:25,496][WARN][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://localhost:9200/"} [2019-09-28T10:31:25,531][INFO][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil} [2019-09-28T10:31:25,534][INFO][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"template"=>"logstash-*", "version"=>50001, "settings"=>{"index.refresh_interval"=>"5s"}, "mappings"=>{"_default_"=>{"_all"=>{"enabled"=>true, "norms"=>false}, "dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"*", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword", "ignore_above"=>256}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date", "include_in_all"=>false}, "@version"=>{"type"=>"keyword", "include_in_all"=>false}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}} [2019-09-28T10:31:25,538][INFO][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["//localhost:9200"]} [2019-09-28T10:31:25,590][INFO][logstash.pipeline] Starting pipeline {"id"=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>500} [2019-09-28T10:31:25,976][INFO][logstash.inputs.beats] Beats inputs: Starting input listener {:address=>"0.0.0.0:5044"} ``` I tried using the flag mentioned here -[https://www.elastic.co/guide/en/logstash/current/shutdown.html](https://www.elastic.co/guide/en/logstash/current/shutdown.html) but i still see this error every now and then ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [September 28, 2019, 6:46pm UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/20 "2019-09-28T18:46:44Z")
I do not understand why the shutdown process would stall when there are 0 in-flight events. ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 29, 2019, 6:55am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/21 "2019-09-29T06:55:34Z")
How do i workaround that ? ---
**Author:** ![Eva](https://avatars.discourse-cdn.com/v4/letter/e/dbc845/32.png) [@Eva](https://discuss.elastic.co/u/Eva)\ **Post date:** [September 30, 2019, 11:08am UTC](https://discuss.elastic.co/t/ship-kafka-logs/200706/22 "2019-09-30T11:08:06Z")
I'm stuck here, kindly provide some suggestion on how to fix these errors. Let me know if you need any further info from me. [Next page](https://discuss.elastic.co/t/ship-kafka-logs/200706.md?page=2)