edenhill/librdkafka — pull request #11 by ZhuYouzhi (2013-06-04 03:06:59+00:00)
Title: Free payload memory in rd_kafka_example when queue is full

Hi, Edenhill. I've learned a lot from discussion with you yesterday.

This patch is try to improve rd_kafka_example a little:

When queue is full due to max_outq_msg_cnt limit, opbuf is allocated but has no chance to deallocate in current code, and this patch fix this problem.

I'm not sure whether behavior in this patch is a little contradictory to RD_KAFKA_OP_F_FREE parameter in semantic. Realization in rd_kafka_produce prevent op_destroy to take effect.
int rd_kafka_produce (rd_kafka_t *rk, char *topic, uint32_t partition,
int msgflags,
char *payload, size_t len) {
rd_kafka_op_t *rko;

if (rk->rk_conf.producer.max_outq_msg_cnt &&
rk->rk_op.rkq_qlen >= rk->rk_conf.producer.max_outq_msg_cnt) {
errno = ENOBUFS;
return -1;
}