private void sendPending ( ) { while ( outstanding . get ( ) < concurrency ) { final TopicQueue queue = pendingTopics . poll ( ) ; if ( queue == null ) { return ; } queue . pending = false ; final int sent = queue . sendBatch ( ) ; // Did we send a whole batch? Then there might be more messages in the queue. Mark as pending again. if ( sent == batchSize ) { queue . pending = true ; pendingTopics . offer ( queue ) ; } } }
Send any pending topics .