{"action":"closed","issue":{"url":"https://api.github.com/repos/elastic/logstash/issues/5333","repository_url":"https://api.github.com/repos/elastic/logstash","labels_url":"https://api.github.com/repos/elastic/logstash/issues/5333/labels{/name}","comments_url":"https://api.github.com/repos/elastic/logstash/issues/5333/comments","events_url":"https://api.github.com/repos/elastic/logstash/issues/5333/events","html_url":"https://github.com/elastic/logstash/issues/5333","id":155899135,"number":5333,"title":"logstash-input-beats blocked,lots of error output","user":{"login":"zxy12","id":4112803,"avatar_url":"https://avatars.githubusercontent.com/u/4112803?v=3","gravatar_id":"","url":"https://api.github.com/users/zxy12","html_url":"https://github.com/zxy12","followers_url":"https://api.github.com/users/zxy12/followers","following_url":"https://api.github.com/users/zxy12/following{/other_user}","gists_url":"https://api.github.com/users/zxy12/gists{/gist_id}","starred_url":"https://api.github.com/users/zxy12/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zxy12/subscriptions","organizations_url":"https://api.github.com/users/zxy12/orgs","repos_url":"https://api.github.com/users/zxy12/repos","events_url":"https://api.github.com/users/zxy12/events{/privacy}","received_events_url":"https://api.github.com/users/zxy12/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2016-05-20T06:57:34Z","updated_at":"2016-05-21T16:19:26Z","closed_at":"2016-05-21T16:19:26Z","body":"my filebeat connect to logstash beat error,  i can't find out where is wrong, my kafka is working well.\r\nplease help me to fix this issue,thanks very much..\r\n\r\n- Version:2.3\r\n- Operating System:centos 6.3\r\n- Config File (if you have sensitive info, please remove it):\r\n```\r\nfilebeat:\r\n  spool_size: 2048\r\n  idle_timeout: 5s\r\n  registry_file: \".registry_file_new\"\r\n  prospectors:\r\n    -\r\n      paths:\r\n          - /data/log/lbs/lbs.log.201*\r\n      encoding: utf-8\r\n      input_type: log\r\n      ignore_older: 1h\r\n      close_older: 24h\r\n      harvester_buffer_size: 16384\r\n      scan_frequency: 10s\r\n      backoff: 1s\r\n      fields:\r\n          log_src: \"lbs\"\r\n    -\r\n      paths:\r\n          - /data/log/ral/ral-worker.log\r\n      input_type: log\r\n      ignore_older: 1h\r\n      close_older: 24h\r\n      harvester_buffer_size: 16384\r\n      scan_frequency: 10s\r\n      backoff: 1s\r\n      fields:\r\n          log_src: \"ral\"\r\n    -\r\n      paths:\r\n          - /data/log/ral/ral-worker.log.wf\r\n      input_type: log\r\n      ignore_older: 1h\r\n      close_older: 24h\r\n      harvester_buffer_size: 16384\r\n      scan_frequency: 10s\r\n      backoff: 1s\r\n      fields:\r\n          log_src: \"ral_wf\"\r\noutput:\r\n\r\n    logstash:\r\n        bulk_max_size: 2048\r\n        hosts:\r\n            - \"slave1:5044\"\r\n            - \"slave1:5045\"\r\n\r\nshipper:\r\n\r\nlogging:\r\n      to_files: true\r\n      level: info\r\n      files:\r\n        path: \"/data/log/filebeat/\"\r\n        name: filebeat.log\r\n        rotateeverybytes: 10485760 # = 10MB\r\n```\r\n```\r\ninput {\r\n  beats {\r\n    port => 5044\r\n    type => \"logs\"\r\n  }\r\n}\r\n\r\noutput {\r\n    if [fields][log_src] == \"lbs\" {\r\n        kafka {\r\n            topic_id => 'odp'\r\n            bootstrap_servers => 'master1:6667,slave1:6667,slave2:6667'\r\n        }\r\n    } else if [fields][log_src] == \"ral\" {\r\n        kafka {\r\n            topic_id => 'ral'\r\n            bootstrap_servers => 'master1:6667,slave1:6667,slave2:6667'\r\n        }\r\n    } else if [fields][log_src] == \"ral_wf\" {\r\n        kafka {\r\n            topic_id => 'ral_wf'\r\n            bootstrap_servers => 'master1:6667,slave1:6667,slave2:6667'\r\n        }\r\n    }\r\n}\r\n```\r\n- Sample Data:\r\n- Steps to Reproduce:\r\n\r\nfilebeat log\r\n```\r\n2016-05-20T14:59:23+08:00 INFO Error publishing events (retrying): EOF\r\n2016-05-20T14:59:29+08:00 INFO Error publishing events (retrying): EOF\r\n2016-05-20T14:59:36+08:00 INFO Error publishing events (retrying): EOF\r\n```\r\n\r\nlogstash.log\r\n```\r\n{:timestamp=>\"2016-05-20T14:53:42.127000+0800\", :message=>#<LogStash::PipelineReporter::Snapshot:0x450ef2da @data={:events_filtered=>5551, :events_consumed=>5551, :worker_count=>4, :inflight_count=>0, :worker_states=>[{:status=>\"dead\", :alive=>false, :index=>0, :inflight_count=>0}, {:status=>\"dead\", :alive=>false, :index=>1, :inflight_count=>0}, {:status=>\"dead\", :alive=>false, :index=>2, :inflight_count=>0}, {:status=>\"dead\", :alive=>false, :index=>3, :inflight_count=>0}], :output_info=>[{:type=>\"kafka\", :config=>{\"topic_id\"=>\"odp\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>567, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:1838327333 [<LogStash::Outputs::Kafka topic_id=>\"odp\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>1}, {:type=>\"kafka\", :config=>{\"topic_id\"=>\"ral\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>4984, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:-572166116 [<LogStash::Outputs::Kafka topic_id=>\"ral\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>0}, {:type=>\"kafka\", :config=>{\"topic_id\"=>\"ral_wf\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>0, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:-1495896771 [<LogStash::Outputs::Kafka topic_id=>\"ral_wf\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>0}], :thread_info=>[], :stalling_threads_info=>[]}>, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:42.484000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:42.484000+0800\", :message=>\"CircuitBreaker::Open\", :name=>\"Beats input\", :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:42.487000+0800\", :message=>\"Beats input: The circuit breaker has detected a slowdown or stall in the pipeline, the input is closing the current connection and rejecting new connection until the pipeline recover.\", :exception=>LogStash::Inputs::BeatsSupport::CircuitBreaker::OpenBreaker, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:42.986000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:43.488000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:43.634000+0800\", :message=>#<LogStash::PipelineReporter::Snapshot:0x6d5b4f2f @data={:events_filtered=>113913, :events_consumed=>113913, :worker_count=>4, :inflight_count=>500, :worker_states=>[{:status=>\"sleep\", :alive=>true, :index=>0, :inflight_count=>125}, {:status=>\"sleep\", :alive=>true, :index=>1, :inflight_count=>125}, {:status=>\"run\", :alive=>true, :index=>2, :inflight_count=>125}, {:status=>\"sleep\", :alive=>true, :index=>3, :inflight_count=>125}], :output_info=>[{:type=>\"kafka\", :config=>{\"topic_id\"=>\"odp\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>22056, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:-1632663525 [<LogStash::Outputs::Kafka topic_id=>\"odp\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>1}, {:type=>\"kafka\", :config=>{\"topic_id\"=>\"ral\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>91725, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:1177379794 [<LogStash::Outputs::Kafka topic_id=>\"ral\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>0}, {:type=>\"kafka\", :config=>{\"topic_id\"=>\"ral_wf\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>0, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:179727203 [<LogStash::Outputs::Kafka topic_id=>\"ral_wf\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>0}], :thread_info=>[{\"thread_id\"=>190, \"name\"=>\"[main]>worker0\", \"plugin\"=>nil, \"backtrace\"=>[\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:114:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:236:in `worker_loop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:205:in `start_workers'\"], \"blocked_on\"=>nil, \"status\"=>\"sleep\", \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}, {\"thread_id\"=>191, \"name\"=>\"[main]>worker1\", \"plugin\"=>nil, \"backtrace\"=>[\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:114:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:236:in `worker_loop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:205:in `start_workers'\"], \"blocked_on\"=>nil, \"status\"=>\"sleep\", \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}, {\"thread_id\"=>192, \"name\"=>\"[main]>worker2\", \"plugin\"=>nil, \"backtrace\"=>[\"[...]/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka/kafka-producer.rb:50:in `call'\", \"[...]/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka/kafka-producer.rb:50:in `send_msg'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-output-kafka-2.0.3/lib/logstash/outputs/kafka.rb:133:in `register'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-codec-json-2.1.3/lib/logstash/codecs/json.rb:42:in `call'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-codec-json-2.1.3/lib/logstash/codecs/json.rb:42:in `encode'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-output-kafka-2.0.3/lib/logstash/outputs/kafka.rb:148:in `receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/outputs/base.rb:83:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/outputs/base.rb:83:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/outputs/base.rb:83:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:130:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:129:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:114:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:236:in `worker_loop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:205:in `start_workers'\"], \"blocked_on\"=>nil, \"status\"=>\"run\", \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka/kafka-producer.rb:50:in `call'\"}, {\"thread_id\"=>193, \"name\"=>\"[main]>worker3\", \"plugin\"=>nil, \"backtrace\"=>[\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:114:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:236:in `worker_loop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:205:in `start_workers'\"], \"blocked_on\"=>nil, \"status\"=>\"sleep\", \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}], :stalling_threads_info=>[{\"thread_id\"=>190, \"name\"=>\"[main]>worker0\", \"plugin\"=>nil, \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}, {\"thread_id\"=>191, \"name\"=>\"[main]>worker1\", \"plugin\"=>nil, \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}, {\"thread_id\"=>192, \"name\"=>\"[main]>worker2\", \"plugin\"=>nil, \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka/kafka-producer.rb:50:in `call'\"}, {\"thread_id\"=>193, \"name\"=>\"[main]>worker3\", \"plugin\"=>nil, \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}]}>, :level=>:warn}\r\n\r\n{:timestamp=>\"2016-05-20T14:53:43.990000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:44.493000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:44.995000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:45.497000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T14:53:45.999000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n```\r\n```\r\n{:timestamp=>\"2016-05-20T15:18:52.935000+0800\", :message=>\"The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information.\", :level=>:error}\r\n{:timestamp=>\"2016-05-20T15:18:56.709000+0800\", :message=>#<LogStash::PipelineReporter::Snapshot:0x629a84e7 @data={:events_filtered=>113913, :events_consumed=>113913, :worker_count=>4, :inflight_count=>500, :worker_states=>[{:status=>\"sleep\", :alive=>true, :index=>0, :inflight_count=>125}, {:status=>\"sleep\", :alive=>true, :index=>1, :inflight_count=>125}, {:status=>\"run\", :alive=>true, :index=>2, :inflight_count=>125}, {:status=>\"sleep\", :alive=>true, :index=>3, :inflight_count=>125}], :output_info=>[{:type=>\"kafka\", :config=>{\"topic_id\"=>\"odp\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>22056, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:-1632663525 [<LogStash::Outputs::Kafka topic_id=>\"odp\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>1}, {:type=>\"kafka\", :config=>{\"topic_id\"=>\"ral\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>91725, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:1177379794 [<LogStash::Outputs::Kafka topic_id=>\"ral\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>0}, {:type=>\"kafka\", :config=>{\"topic_id\"=>\"ral_wf\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>0, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:179727203 [<LogStash::Outputs::Kafka topic_id=>\"ral_wf\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>0}], :thread_info=>[{\"thread_id\"=>190, \"name\"=>\"[main]>worker0\", \"plugin\"=>nil, \"backtrace\"=>[\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:114:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:236:in `worker_loop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:205:in `start_workers'\"], \"blocked_on\"=>nil, \"status\"=>\"sleep\", \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}, {\"thread_id\"=>191, \"name\"=>\"[main]>worker1\", \"plugin\"=>nil, \"backtrace\"=>[\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:114:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:236:in `worker_loop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:205:in `start_workers'\"], \"blocked_on\"=>nil, \"status\"=>\"sleep\", \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}, {\"thread_id\"=>192, \"name\"=>\"[main]>worker2\", \"plugin\"=>nil, \"backtrace\"=>[\"[...]/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka/kafka-producer.rb:50:in `call'\", \"[...]/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka/kafka-producer.rb:50:in `send_msg'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-output-kafka-2.0.3/lib/logstash/outputs/kafka.rb:133:in `register'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-codec-json-2.1.3/lib/logstash/codecs/json.rb:42:in `call'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-codec-json-2.1.3/lib/logstash/codecs/json.rb:42:in `encode'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-output-kafka-2.0.3/lib/logstash/outputs/kafka.rb:148:in `receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/outputs/base.rb:83:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/outputs/base.rb:83:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/outputs/base.rb:83:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:130:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:129:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:114:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:236:in `worker_loop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:205:in `start_workers'\"], \"blocked_on\"=>nil, \"status\"=>\"run\", \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka/kafka-producer.rb:50:in `call'\"}, {\"thread_id\"=>193, \"name\"=>\"[main]>worker3\", \"plugin\"=>nil, \"backtrace\"=>[\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `worker_multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:114:in `multi_receive'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `each'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:305:in `output_batch'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:236:in `worker_loop'\", \"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/pipeline.rb:205:in `start_workers'\"], \"blocked_on\"=>nil, \"status\"=>\"sleep\", \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}], :stalling_threads_info=>[{\"thread_id\"=>190, \"name\"=>\"[main]>worker0\", \"plugin\"=>nil, \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}, {\"thread_id\"=>191, \"name\"=>\"[main]>worker1\", \"plugin\"=>nil, \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}, {\"thread_id\"=>192, \"name\"=>\"[main]>worker2\", \"plugin\"=>nil, \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/jruby-kafka-1.5.0-java/lib/jruby-kafka/kafka-producer.rb:50:in `call'\"}, {\"thread_id\"=>193, \"name\"=>\"[main]>worker3\", \"plugin\"=>nil, \"current_call\"=>\"[...]/vendor/bundle/jruby/1.9/gems/logstash-core-2.3.0-java/lib/logstash/output_delegator.rb:128:in `pop'\"}]}>, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T15:18:57.127000+0800\", :message=>#<LogStash::PipelineReporter::Snapshot:0x1e11514e @data={:events_filtered=>5551, :events_consumed=>5551, :worker_count=>4, :inflight_count=>0, :worker_states=>[{:status=>\"dead\", :alive=>false, :index=>0, :inflight_count=>0}, {:status=>\"dead\", :alive=>false, :index=>1, :inflight_count=>0}, {:status=>\"dead\", :alive=>false, :index=>2, :inflight_count=>0}, {:status=>\"dead\", :alive=>false, :index=>3, :inflight_count=>0}], :output_info=>[{:type=>\"kafka\", :config=>{\"topic_id\"=>\"odp\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>567, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:1838327333 [<LogStash::Outputs::Kafka topic_id=>\"odp\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>1}, {:type=>\"kafka\", :config=>{\"topic_id\"=>\"ral\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>4984, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:-572166116 [<LogStash::Outputs::Kafka topic_id=>\"ral\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>0}, {:type=>\"kafka\", :config=>{\"topic_id\"=>\"ral_wf\", \"bootstrap_servers\"=>\"master1:6667,slave1:6667,slave2:6667\"}, :is_multi_worker=>false, :events_received=>0, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:-1495896771 [<LogStash::Outputs::Kafka topic_id=>\"ral_wf\", bootstrap_servers=>\"master1:6667,slave1:6667,slave2:6667\", codec=><LogStash::Codecs::JSON charset=>\"UTF-8\">, workers=>1, key_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", value_serializer=>\"org.apache.kafka.common.serialization.StringSerializer\", acks=>\"1\", buffer_memory=>33554432, compression_type=>\"none\", retries=>0, batch_size=>16384, linger_ms=>0, max_request_size=>1048576, receive_buffer_bytes=>32768, send_buffer_bytes=>131072, timeout_ms=>30000, block_on_buffer_full=>true, metadata_fetch_timeout_ms=>60000, metadata_max_age_ms=>300000, reconnect_backoff_ms=>10, retry_backoff_ms=>100>]>, :busy_workers=>0}], :thread_info=>[], :stalling_threads_info=>[]}>, :level=>:warn}\r\n```\r\n```\r\n{:timestamp=>\"2016-05-20T15:30:29.452000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T15:30:29.956000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T15:30:30.459000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T15:30:30.961000+0800\", :message=>\"Beats input: the pipeline is blocked, temporary refusing new connection.\", :reconnect_backoff_sleep=>0.5, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T15:31:34.403000+0800\", :message=>\"CircuitBreaker::rescuing exceptions\", :name=>\"Beats input\", :exception=>LogStash::Inputs::Beats::InsertingToQueueTakeTooLong, :level=>:warn}\r\n{:timestamp=>\"2016-05-20T15:31:34.405000+0800\", :message=>\"Beats input: The circuit breaker has detected a slowdown or stall in the pipeline, the input is closing the current connection and rejecting new connection until the pipeline recover.\", :exception=>LogStash::Inputs::BeatsSupport::CircuitBreaker::HalfOpenBreaker, :level=>:warn}\r\n```\r\n\r\nseem input workers are dead, why?"}}
{"actor":{"display_login":"zxy12"}}