Sleeping in the consumer-main-loop allowed?

[SoulfreezerXP]

Hello,

I have a question focussing thread-sleeping in the consumer-main-loop. My consumer-loop
runs in an own thread which periodically calls the consume-function in a while-loop. The msg will be transferred to my own
callback which transfers the msg to some global queue. Other threads are working at this queue. All is working great, but If my global queue is running full, I need to stop the consumer-thread. I am using C++17 and I use internally a std::condition_variable
which will force the consumer-loop-thread to a passive sleep. My question is now: Is this a good solution, because I have read, even when I will not call the consume()-function, a separated kafka-thread will still fetching msg from the broker and handle other things. Maybe I should call _consumer->pause() before going sleep and after awakening the resume()-function? But what happens during rebalance-callbacks() and so on? Is it in general not ok, to sleep in the consumer-loop? It would be really great if you could point me in the right direction to build a robust consumer-thread. Thank you very much!

Actual implementation:

        //Consumer-Loop in an own thread
        while ( !this->_shouldShutdown )
        {
            RdKafka::Message *msg = _consumer->consume( 1000 );

            //Here a thread-sleep can occur when queue is full. The sleep occurs BEFORE
            //the callback has put the msg into the queue, so it will be processed AFTER wakeup. 
            (*_pConsumerCB)( *msg, NULL ); 

            delete msg;
        }

Possible solution with pause()/resume ?

       //Consumer-Loop in an own thread
        while ( !this->_shouldShutdown )
        {
            RdKafka::Message *msg = _consumer->consume( 1000 );
           
           //When queue is full -> call pause(), sleep..., and when awake then call resume()
           //Is the pending msg here still valid AFTER (wakeup AND resume) 
           //and can be processed correctly?
            (*_pConsumerCB)( *msg, NULL ); 
             delete msg;
        }

Maybe a solution without sleeping ?

        //Consumer-Loop in an own thread
        while ( !this->_shouldShutdown )
        {
            if ( _consumer->isPaused) //Is there a flag to check this?
              _consumer->resume(...)

            RdKafka::Message *msg = _consumer->consume( 1000 );
                          
           //CB returns false when queue is full
           //Is the msg still valid when queue becomes free and pause() was called before?
           while (!(*_pConsumerCB)( *msg, NULL ) ) 
           {
                if ( !_consumer->isPaused) //Is there a flag to check this?
                   _consumer->pause(...);

                _consumer->consume( 2000 ); //Still necessary ?               
            }      
      
            delete msg; //Msg still valid?
        }

[edenhill]

You must call poll/consume() at least every max.poll.interval.ms or your consumer will get kicked out of the group. Preferably much more often than that.

pause()ing when your workers are congested and then resume()ing again when there is room will work, but do note that pause() will purge the internal pre-fetch queue of messages, which defaults to up to 64 megs. So continually pausing and resuming may induce a high network load as messages are re-fetched again.
You can limit the pre-fetch queue by queued.max.messages.kbytes.

[SoulfreezerXP]

Thank you very much for your time to help me!

If I understand your answer correct, then I have two options if my queue is full, right?
Option 1 (the middle of my examples) is to pause->sleep->resume in the consumer-thread
without calling continously consume(). And the option 2 would be to completly
avoid sleeping and calling continously the poll-function instead of the
consume-function when the queue went full. I still have 3 questions to this.

1. Is in option1 the msg-pointer still valid after purging the pre-fetch queue, so that I can use and delete that msg after resuming?

2. What will happen in option1 and option2, when the pre-fetch queue also went full?

3. Which option would be your advise?