High concurrency usage

[andrerigon]

Hi,

I'm writing a nginx module, which parses a message and send it to kafka using librdkafka. But under high level of concurrency, the api locks e and says it cannot create more threads to send the message. Is there a way to overcome this, by using it in a synchronous way, for example?

thanks in advance,

andre

[edenhill]

Hi Andre,

what version/branch is this? 0.7 or 0.8/master?

[edenhill]

How many rd_kafka_t objects are you creating, with how many topics, and how many brokers are available in the cluster?
Also; are you creating and destroy multiple rd_kafka_t objects? (i.e., shortlived)

[andrerigon]

Hi Magnus,

I'm using version 0.8/master.
I create a rd_kafka_t per reques:

rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, &conf,
errstr, sizeof(errstr));

than send the msg and wait:

rd_kafka_topic_t *rkt = rd_kafka_topic_new(rk, topic, &topic_conf);

rd_kafka_produce(rkt, partition,
RD_KAFKA_MSG_F_FREE, msg, strlen(msg),
"nginx", 5,
NULL);

rd_kafka_poll(rk, 0);

and finally:

rd_kafka_destroy(rk);

Thanks!

[edenhill]

Thanks for clarifying your setup.

My guess is that the broker threads created by rdkafka arent properly terminated on rd_kafka_destroy() and that is clearly a bug that I will look into.

If possible I suggest that you try to hang on to the rd_kafka_t object and reuse it in subsequent rd_kafka_produce() calls to amortize the startup cost of connecting to the broker(s) and fetching metadata for each produced message. Not only will it avoid the above bug but it will provide a far more performant and resource frugal setup.
Would that be possible in your case?

[edenhill]

I'm working on a fix, should be available end of this week.

[edenhill]

Andre, could you please pull the latest master and give it another go?
This issue should now be fixed and it would be great if you could verify it in your setup.
Reopen this issue if the problem remains.

[andrerigon]

Hi,

the problem is gone with the fix.

thanks!