Producer buffer available bytes constantly decreasing

[Jan Algermissen]

Hi,

I am writing to a Kafka Topic from within a Scala/Akka application using the 
new Producer (0.8.2.1).

While writing messages to the queue (at a very reasonable rate of a couple of 
messages per second max) the available buffer is constantly decreasing until 
the producer finally throws an exception saing the buffer is exhausted.

There must be something I do not understand that I am doing wrong, can anyone 
provde a clue?

Here is my config:

 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
 ProducerConfig.RETRIES_CONFIG -> "0",
 ProducerConfig.ACKS_CONFIG -> "1",
 ProducerConfig.COMPRESSION_TYPE_CONFIG -> "none",
 ProducerConfig.TIMEOUT_CONFIG -> new Integer(30000),
// ProducerConfig.BATCH_SIZE_CONFIG -> new Integer(16384),
 ProducerConfig.BATCH_SIZE_CONFIG -> new Integer(10),
 ProducerConfig.BUFFER_MEMORY_CONFIG -> new Integer(66554432),
 ProducerConfig.BLOCK_ON_BUFFER_FULL_CONFIG -> new java.lang.Boolean(false),
 ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG -> 
"org.apache.kafka.common.serialization.StringSerializer",
 ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG -> 
"org.apache.kafka.common.serialization.StringSerializer"


And my send code:

kafkaProducer.send(new ProducerRecord[String,String](topic, key, data),new 
Callback {
  def onCompletion(recordMetadata: RecordMetadata, e: Exception):Unit = {
    if(e != null) {
      logger.error(s"Could not send $data",e)
    }
    logger.info("The offset of the record we just sent is: " + 
recordMetadata.offset())
    ()
  }

})


I am using the metrics() member to periodically look at 
"buffer-available-bytes" and I see it is constantly decreasing over time as 
messages are being sent.


Jan