ogibayashi

@@ -114,6 +116,7 @@ public class FlinkKafkaConsumer09<T> extends FlinkKafkaConsumerBase<T> {
 	/** If the consumer doesn't have a Kafka partition assigned at runtime, it'll block on this waitThread **/
 	private transient Thread waitThread;
 
+        private final ReentrantLock lock = new ReentrantLock(true);
 
 	// ------------------------------------------------------------------------
 
@@ -388,9 +391,12 @@ public class FlinkKafkaConsumer09<T> extends FlinkKafkaConsumerBase<T> {

ogibayashi

2016-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"
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/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'
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:123:in `each'
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/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'
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/plugin/buf_file.rb:71:in `open'
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:120:in `msgpack_each'
  2016-10-06 17:10:27 +0900 [w

ogibayashi

---
hostPort: localhost:{{ item.jmx_port }}
lowercaseOutputName: true
lowercaseOutputLabelNames: false
whitelistObjectNames:
  - "java.lang:*"
  - "org.apache.flink.metrics:*"
rules:
  - pattern: "^java.lang.*"
# Kafka source