Librdkafka: Produce failed: Local: Queue full

[abhit011]

The entire setup is on Windows.
My Kafka Cluster has 6 brokers, 4 on one machine and 2 on other machine.
It has two Kafka Topics with partition size 50 each, and replication factor of 3.

My partition logic selection: like for each message (its unique ID % 50) , and then calling Kafka producer API to route a specific message to a particular topic partition .
RdKafka::ErrorCode resp = m_kafkaProducer->produce(m_kafkaTopic,(unique ID % 50),RdKafka::Producer::RK_MSG_COPY /* Copy payload */,ptr,size,&partationKey,NULL);

I am using various process to produce messages to Kafka to these 2 topics.
On an average there are around 48 threads in each process thats calling m_kafkaProducer->produce(m_kafkaTopic,partition,RdKafka::Producer::RK_MSG_COPY /* Copy payload */,ptr,size,&partationKey,NULL);

Its still under POC stage, so currently there are only 2 process thats producing to Kafka.

Each produced message is of 96 bytes each, and max traffic right now is 1000 messages / sec. I have aim to reach close to 800k-1M messages per sec from various processes to Kafka Cluster..

Of late i experienced error like when ever my process tried to produce it gave error saying "Produce failed: Local: Queue full" where are at the same time with in the same process after I got this Queue Full error, there were few other messages that got produced successfully ? why this is happneing.

My Kafka Producer configuration
```m_KafkaConf->set("metadata.broker.list", m_brokersAddress, errstr); m_KafkaConf->set("dr_cb", ex_dr_cb, errstr);
if (m_ex_event_cb == NULL)
return -1;
m_KafkaConf->set("event_cb", m_ex_event_cb, errstr);
t_KafkaConf->set("produce.offset.report", "true", errstr);
t_KafkaConf->set("request.required.acks", "1", errstr);

And I have just one KafkaConsumer, which is going to consume all the kafka messages. Currently I have restricted that to just 8 threads. I mean 8 threads in my kafka consumer are using to consumer messages and then doing the actions. 
Kafka Consumer setting

t_KafkaConf->set("auto.offset.reset", "latest", errstr);
t_KafkaConf->set("enable.auto.commit", "true", errstr);
t_KafkaConf->set("auto.commit.interval.ms", "1000", errstr);
m_KafkaConf->set("rebalance_cb", m_ex_rebalance_cb, errstr);
m_KafkaConf->set("metadata.broker.list", m_brokersAddress, errstr);

I am using 0.9.2 librdkafka for Producer and Consumer. 

And my server.properties is

broker.id=0
port:9093
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
offsets.retention.minutes=360
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.minutes=360
log.segment.bytes=52428800
log.retention.check.interval.ms=300000
log.cleaner.enable=true
log.cleanup.policy=delete
log.cleaner.min.cleanable.ratio=0.5
log.cleaner.backoff.ms=15000
log.segment.delete.delay.ms=6000
auto.create.topics.enable=false
```
Why these Queue Full came I feel on few partitions, where the traffic itself right now is not huge?

Abhi

[abhit011]

Any pointers Magnus @edenhill ?

Abhi

[abhit011]

And I have not done anything to queue.buffering.max.messages, batch.num.messages and queue.buffering.max.messages settings at my producer side ? Then what could be the reason ?
And this Local: Queue Full , occurs in which stage of Produce()
In-flight mode ? In-ack res ? or dr_db() ?
Please pass some pointers ? I am literally stuck on it :(

Abhi

[abhit011]

@edenhill its little SOS situation :(

[abhit011]

Can any one please guide me what possible steps I need to consider ?

Abhi

[edenhill]

Sorry about the delay.

produce() will return QUEUE_FULL when the internal producer queue gets full (queue.buffering.max.messages, or ...bytes).
The quee does not only account for messages being queued for transmission to the broker but also for acknowledgements - delivery reports - waiting to be polled by your application.
So if you register a delivery report callback you must also call rd_kafka_poll() at regular intervals to serve any queued delivery report callbacks.

This is typically done from your main loop or through its own thread.

If you are sure that you will have a constant flow of messages to produce you can probably get away with calling poll(0) right after each produce() call, this will not serve callback for the message just produce()d, but for earlier messages that have no succeeded (or failed) transmission.

[abhit011]

@edenhill thanks for the comment, let me write how I am calling poll().

In my Main(), I am creating single Kafka Produce instance and then calling the m_Kafkainstance->Produce() API from 48 threads with in a process. Now there is another thread which is calling poll like

while (!CThread::currentThread().isInterrupted()) { m_Ptr->getKafkaProducerInstance()->poll(1000); }

Is this the correct approach ?

I thought that poll(0) is very expensive operation and should be avoided.

Now lets say I start my process and there there 4 calls to produce(), like

m_Kafkainstance->Produce(); // producing 1, 
m_Kafkainstance->poll(0);

m_Kafkainstance->Produce(); // producing 2, 
m_Kafkainstance->poll(0);

m_Kafkainstance->Produce(); // producing 3, 
m_Kafkainstance->poll(0);

m_Kafkainstance->Produce(); // producing 4, 
m_Kafkainstance->poll(0);

Will I get the delievery_cb for 1,2,3,4 ? Will poll(0) add any overhead ?
The flow of the messages initially will start from 1 and then it will gets picked up to its max , can go to 10k too. In that case ?
Abhi

[edenhill]

Your current approach with a dedicated poll thread is good, it is better than the poll(0) approach (and no, poll(0) isnt expensive, it is just a lock+checkptr+unlock).

Can you register a stats callback and set statistics.interval.ms=5000 or so, and provide a stats dump when you've reached QUEUE_FULL? This way we can see in what queues things are held up.

[abhit011]

Okie, I need to reproduce it, give me some time. I need to see if again it gets reproduced or not. I will let u know ..

Abhi

[abhit011]

STATSDUMP.txt
After this time stamp started getting queue full error..
Since my process is doing too much oepration , i had given statistics.interval.ms to 20000

if (m_KafkaConf->set("statistics.interval.ms", "20000", errstr) 
                            != RdKafka::Conf::CONF_OK){
            kafka_logger1->critic() << "Failed to create Kafka Producer for statistics.interval.ms";
            return -1;
        }

Does StatsDump give any info ?

Abhi

[edenhill]

{
  "name": "rdkafka#producer-1",
  "type": "producer",
  "ts": 2507587968000,
  "time": 1484662310,
  "replyq": 0,
  "msg_cnt": 81,
  "msg_size": 7776,
  "msg_max": 100000,
  "msg_size_max": -198967296,
  "simple_cnt": 0,
...

That msg_size_max doesn't look right, will investigate

[abhit011]

Do i need to add more stuff, please let me know I will also see

Abhi

[edenhill]

What Windows OS version are you on, and platform?

[abhit011]

Windows 2012 server

[edenhill]

32 or 64 bit librdkafka?

[abhit011]

I did not change anything related to that , it feel its 32 bit by default

[edenhill]

Okay, that makes sense, this is probably a wrapping issue with the code assuming size_t to be 64-bit.
I'll look into it.

Out of curiosity: why 32-bit and not 64? Is 32-bits still typical for Windows apps?

[edenhill]

Could you try setting the producer property queue.buffering.max.kbytes to 1000000?

[abhit011]

Because other static lib that I am using in current system (built by few guys few years ago were in 32 bit only) , so i thought let me build in 32 bit only thats was the reason..Any way I will try your suggestion now and keep you posted about the results..
Thanks
Abhi

[abhit011]

m_KafkaConf->set("queue.buffering.max.kbytes", "1000000", errstr);
But ran on local machine, and atleast what you pointed "msg_size_max":1024000000, is coming like this..
Which is exactly 1024 x 1000000 , where prev it was coming -negative
But i need to test this on the server to see does this solve the Local:Queue Full error or not..
I need to wait for few hours to see it happen.

Abhi

[abhit011]

STATSDMP.txt
After making your suggested change, ran it for around 25 - 30 minutes, that Local Queue Full error did not occur. But i need to see it in future if it comes. Also I am attaching fresh stats file after change for your reference
I need to understand few things

name": "rdkafka#producer-1", "type": "producer", "ts":2586628906000, "time":1484741351, "replyq":394, "msg_cnt":6074, "msg_size":583104, "msg_max":100000, "msg_size_max":1024000000,

What replyq ? msg_cnt ? and msg_size signifies, i cna see in stats that its increasing and decreasing . I need to understand that some problem should not occur again

Thanks
Abhi

[edenhill]

That looks good!

• msg_cnt is the number of messages currently in librdkafka queues
• msg_size is the sum of those messages' payloads
• replyq is the number of ops waiting for your application to poll().

An op is a generic event inside librdkafka, in the producer case it is typically a delivery report callback, or a stats_cb.

[edenhill]

@shatilov-diman I guess you mean issue #1980, which has already been fixed.

[shatilov-diman]

@shatilov-diman I guess you mean issue #1980, which has already been fixed.
Yes, thank you

[ALTAMASH93]

Guys, I see the discussion is about Windows. Is it possible to setup librdkafka on Windows? I've been trying to, but due to the (like pthreads and all), I havent been able to set it up on windows? Is it still possible? Any leads will be of great help.

[ALTAMASH93]

@abhit011 @edenhill can you please suggest?

[edenhill]

@ALTAMASH93 Reference the librdkafka.redist NuGet package in your Visual Studio project.

Example: https://github.com/edenhill/kafkacat/blob/master/win32/kafkacat.vcxproj

[abhit011]

You want to compile librdkafka on Windows ? I did year and half ago and have been using since then... Didn't update after that because I got busy in writing some other components.. but please let me know what help you need..

[SupperSu]

Sorry about the delay.

produce() will return QUEUE_FULL when the internal producer queue gets full (queue.buffering.max.messages, or ...bytes).
The quee does not only account for messages being queued for transmission to the broker but also for acknowledgements - delivery reports - waiting to be polled by your application.
So if you register a delivery report callback you must also call rd_kafka_poll() at regular intervals to serve any queued delivery report callbacks.

This is typically done from your main loop or through its own thread.

If you are sure that you will have a constant flow of messages to produce you can probably get away with calling poll(0) right after each produce() call, this will not serve callback for the message just produce()d, but for earlier messages that have no succeeded (or failed) transmission.

That is helpful! Thank you for your replay!
I am wondering does multiple producers share the same producer internal queue ? or each producer has own separate one ?

[edenhill]

Producer instances are completely separated, they do not share anything.