hi,

we noticed number of async threads keeps on growing (with or with out any
load). This might be triggered by some flacky zookeeper where kafka brokers
leave and join back. I am thinking of updating ProducerPool to not create
new threads without closing the existing ones
(ProducerPool.java:85<https://github.com/apache/kafka/blob/0.7.2/core/src/main/scala/kafka/producer/ProducerPool.scala#L85>).
Would that be a good fix?

thanks.
Raghu.

e.g. log:
[...]
12/11/13 20:42:47 INFO producer.ProducerPool: Creating async producer for
broker id = 0 at 10.35.26.118:31222
12/11/13 20:44:08 INFO producer.ProducerPool: Creating async producer for
broker id = 3 at 10.35.46.114:31609
12/11/13 20:48:37 INFO producer.ProducerPool: Creating async producer for
broker id = 3 at 10.35.46.114:31609
12/11/13 20:49:17 INFO producer.ProducerPool: Creating async producer for
broker id = 3 at 10.35.46.114:31609
12/11/13 20:49:19 INFO producer.ProducerPool: Creating async producer for
broker id = 0 at 10.35.26.118:31222
12/11/13 20:50:25 INFO producer.ProducerPool: Creating async producer for
broker id = 3 at 10.35.46.114:31609
12/11/13 20:50:38 INFO producer.ProducerPool: Creating async producer for
broker id = 5 at 10.35.20.117:31627
12/11/13 20:50:56 INFO producer.ProducerPool: Creating async producer for
broker id = 3 at 10.35.46.114:31609
12/11/13 20:51:42 INFO producer.ProducerPool: Creating async producer for
broker id = 0 at 10.35.26.118:31222
[...]

debug log:

12/11/13 21:00:38 TRACE
producer.ZKBrokerPartitionInfo$BrokerTopicsListener: Watcher fired for
path: /brokers/topics/test_category with change [3, 2, 1, 0, 7, 6, 5, 4, 9,
8]
12/11/13 21:00:38 DEBUG
producer.ZKBrokerPartitionInfo$BrokerTopicsListener: [BrokerTopicsListener]
List of brokers changed at /brokers/topics/test_category       Currently
registered  list of brokers -> [3, 2, 1, 0, 7, 6, 5, 4, 9, 8] for topic ->
test_category
12/11/13 21:00:38 DEBUG
producer.ZKBrokerPartitionInfo$BrokerTopicsListener: [BrokerTopicsListener]
Old list of brokers: TreeSet(0, 1, 2, 3, 4, 5, 6, 7, 8)
12/11/13 21:00:38 DEBUG
producer.ZKBrokerPartitionInfo$BrokerTopicsListener: [BrokerTopicsListener]
Currently registered list of brokers for topic: test_category are Buffer(3,
2, 1, 0, 7, 6, 5, 4, 9, 8)
12/11/13 21:00:38 DEBUG
producer.ZKBrokerPartitionInfo$BrokerTopicsListener: [BrokerTopicsListener]
Unregistered list of brokers for topic: test_category are TreeSet(0-0, 1-0,
2-0, 3-0, 4-0, 5-0, 6-0, 7-0, 8-0)
12/11/13 21:00:38 DEBUG
producer.ZKBrokerPartitionInfo$BrokerTopicsListener: [BrokerTopicsListener]
List of broker partitions for topic: test_category are TreeSet(0-0, 1-0,
2-0, 3-0, 4-0, 5-0, 6-0, 7-0, 8-0, 9-0)
[...]