hi, community:
The implements of closable blocking queue[1] use a fair lock ReentrantLock to guarantee thread-safe. The changes below may be help:
1. replace ReentrantLock with ReentrantReadWriteLock to improve concurrency: in fair mode, the ReentrantReadWriteLock has better performance than ReentrantLock;[2]
2. replace singnalAll() with signal() to reduce the thread scheduling: signal() is safe and reasonable, since all the threads which waiting on the nonEmpty Condition wish to take element from queue, and at most one thread can get element from queue.
jira:
https://issues.apache.org/jira/browse/FLINK-19089[1]
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java[2]
http://isuru-perera.blogspot.com/2016/05/benchmarking-java-locks-with-counters.html--
Best,
kui
Locked
