{"action":"created","issue":{"url":"https://api.github.com/repos/fluent/fluent-plugin-kafka/issues/80","repository_url":"https://api.github.com/repos/fluent/fluent-plugin-kafka","labels_url":"https://api.github.com/repos/fluent/fluent-plugin-kafka/issues/80/labels{/name}","comments_url":"https://api.github.com/repos/fluent/fluent-plugin-kafka/issues/80/comments","events_url":"https://api.github.com/repos/fluent/fluent-plugin-kafka/issues/80/events","html_url":"https://github.com/fluent/fluent-plugin-kafka/issues/80","id":181566887,"number":80,"title":"Flush error with invalid fluentd record","user":{"login":"ogibayashi","id":3626322,"avatar_url":"https://avatars.githubusercontent.com/u/3626322?v=3","gravatar_id":"","url":"https://api.github.com/users/ogibayashi","html_url":"https://github.com/ogibayashi","followers_url":"https://api.github.com/users/ogibayashi/followers","following_url":"https://api.github.com/users/ogibayashi/following{/other_user}","gists_url":"https://api.github.com/users/ogibayashi/gists{/gist_id}","starred_url":"https://api.github.com/users/ogibayashi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ogibayashi/subscriptions","organizations_url":"https://api.github.com/users/ogibayashi/orgs","repos_url":"https://api.github.com/users/ogibayashi/repos","events_url":"https://api.github.com/users/ogibayashi/events{/privacy}","received_events_url":"https://api.github.com/users/ogibayashi/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2016-10-07T01:40:44Z","updated_at":"2016-10-11T05:08:00Z","closed_at":null,"body":"Hello,\r\n\r\nLast day, out_kafka_buffered in my fluentd cluster had the following error.\r\n\r\n```\r\n2016-10-06 17:10:27 +0900 [warn]: temporarily failed to flush the buffer. next_retry=2016-10-06 17:10:28 +0900 error_class=\"NoMethodError\" error=\"undefined method `[]=' for nil:NilClass\" plugin_id=\"to_kafka\"\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluent-plugin-kafka-0.2.2/lib/fluent/plugin/out_kafka_buffered.rb:215:in `block in write'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:123:in `each'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:123:in `block in msgpack_each'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/plugin/buf_file.rb:71:in `open'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:120:in `msgpack_each'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluent-plugin-kafka-0.2.2/lib/fluent/plugin/out_kafka_buffered.rb:210:in `write'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:354:in `write_chunk'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:333:in `pop'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/output.rb:338:in `try_flush'\r\n  2016-10-06 17:10:27 +0900 [warn]: /home/myuser/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/output.rb:149:in `run'\r\n2016-10-06 17:10:28 +0900 [warn]: Send exception occurred: undefined method `[]=' for nil:NilClass\r\n2016-10-06 17:10:28 +0900 [info]: initialized kafka producer: kafka  \r\n```\r\n\r\nFrom the errors from another fluentd which receives the same log, I realized that some of our client sent records with `nil`.\r\n\r\n```\r\n2016-10-06 17:10:08 +0900 [warn]: dump an error event: error_class=NoMethodError error=\"undefined method `/' for \\\"POST\\\":String\" tag=\"sometag\" time=1475741408 record={\"time\"=>\"POST\", \"record\"=>nil}\r\n```\r\n\r\nI know that this is the issue of the sender, but since our fluentd cluster aggregates logs from so many different client, this can happen. And once this happens, other normal logs will also get stuck in the buffer. So, it would be great if out_kafka_buffered could check such invalid record and drop them so that other normal logs will not be affected.\r\n"},"comment":{"url":"https://api.github.com/repos/fluent/fluent-plugin-kafka/issues/comments/252815136","html_url":"https://github.com/fluent/fluent-plugin-kafka/issues/80#issuecomment-252815136","issue_url":"https://api.github.com/repos/fluent/fluent-plugin-kafka/issues/80","id":252815136,"user":{"login":"repeatedly","id":16928,"avatar_url":"https://avatars.githubusercontent.com/u/16928?v=3","gravatar_id":"","url":"https://api.github.com/users/repeatedly","html_url":"https://github.com/repeatedly","followers_url":"https://api.github.com/users/repeatedly/followers","following_url":"https://api.github.com/users/repeatedly/following{/other_user}","gists_url":"https://api.github.com/users/repeatedly/gists{/gist_id}","starred_url":"https://api.github.com/users/repeatedly/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/repeatedly/subscriptions","organizations_url":"https://api.github.com/users/repeatedly/orgs","repos_url":"https://api.github.com/users/repeatedly/repos","events_url":"https://api.github.com/users/repeatedly/events{/privacy}","received_events_url":"https://api.github.com/users/repeatedly/received_events","type":"User","site_admin":false},"created_at":"2016-10-11T05:08:00Z","updated_at":"2016-10-11T05:08:00Z","body":"Hmm... catching invalid records and logging it in error is one idea. I will do it."}}
{"actor":{"display_login":"repeatedly"}}