If the consumer thread pool hasn't been created, create one here.
431254
private synchronized void createConsumerThreadPool() {
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.entry(tc, "createConsumerThreadPool");
        if (_consumerThreadPool == null) {
            // 214163 Add a threadpool listener to associate threads in trace
            // with the messaging engine.
            _consumerThreadPool = new ThreadPool(
                            "Consumer " + getMessagingEngineName(),
                            0,
                            _customProperties.get_max_consumer_threadpool_size(),
                            new ThreadPoolListener[] { new ThreadPoolListenerImpl(this) });
            _consumerThreadPool
                            .setRequestBufferExpansionLimit(Integer.MAX_VALUE);
        }
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.exit(tc, "createConsumerThreadPool");
    }
1013845
def _start_consumer(self, consumer):
        
        thread = threading.Thread(target=consumer.start,
                                  args=(self._connection,))
        thread.daemon = True
        thread.start()
1168754
def _create_consumer(self):
        """"""
        if not self.closed:
            try:
                self.logger.debug("Creating new kafka consumer using brokers: " +
                                   str(self.settings['KAFKA_HOSTS']) + ' and topic ' +
                                   self.settings['KAFKA_TOPIC_PREFIX'] +
                                   ".outbound_firehose")

                return KafkaConsumer(
                    self.settings['KAFKA_TOPIC_PREFIX'] + ".outbound_firehose",
                    group_id=None,
                    bootstrap_servers=self.settings['KAFKA_HOSTS'],
                    consumer_timeout_ms=self.settings['KAFKA_CONSUMER_TIMEOUT'],
                    auto_offset_reset=self.settings['KAFKA_CONSUMER_AUTO_OFFSET_RESET'],
                    auto_commit_interval_ms=self.settings['KAFKA_CONSUMER_COMMIT_INTERVAL_MS'],
                    enable_auto_commit=self.settings['KAFKA_CONSUMER_AUTO_COMMIT_ENABLE'],
                    max_partition_fetch_bytes=self.settings['KAFKA_CONSUMER_FETCH_MESSAGE_MAX_BYTES'])
            except KeyError as e:
                self.logger.error('Missing setting named ' + str(e),
                                   {'ex': traceback.format_exc()})
            except:
                self.logger.error("Couldn't initialize kafka consumer for topic",
                                   {'ex': traceback.format_exc()})
                raise
123179
def add_thread(self, checker, end_callback, source_code, parent):
        """"""
        parent_id = id(parent)
        thread = AnalysisThread(self, checker, source_code)
        self.end_callbacks[id(thread)] = end_callback
        self.pending_threads.append((thread, parent_id))
        logger.debug("Added thread %r to queue" % thread)
        QTimer.singleShot(50, self.update_queue)
360814
public static <T> Consumer<T> monitor(Consumer<T> consumer, AtomicLong calls) {
        return new MonitoringConsumer<T>(consumer, calls);
    }
728051
private void growThreadPool() {

        if (m_currentThreadCount < m_maxThreadCount) {
            // if maximum number is not reached grow the thread pool
            synchronized (m_nextRunnableLock) {
                m_workers[m_currentThreadCount] = new CmsSchedulerThread(
                    this,
                    m_threadGroup,
                    m_threadNamePrefix + m_currentThreadCount,
                    m_threadPriority,
                    m_makeThreadsDaemons);
                m_workers[m_currentThreadCount].start();
                if (m_inheritLoader) {
                    m_workers[m_currentThreadCount].setContextClassLoader(
                        Thread.currentThread().getContextClassLoader());
                }
                // increas the current size
                m_currentThreadCount++;
                // notify the waiting threads
                m_nextRunnableLock.notifyAll();
            }
        }
    }
1061125
def _create_process_thread(self):
        
        thread = threading.Thread(target=self._process_data_events)
        thread.setDaemon(True)
        thread.start()
102696
def thread_pool():
        
        if not LocalImage._thread_pool:
            logger.info("Starting LocalImage threadpool")
            LocalImage._thread_pool = concurrent.futures.ThreadPoolExecutor(
                thread_name_prefix="Renderer")
        return LocalImage._thread_pool
417661
@Override
    public Server create(ThreadPool threadPool) {
        return threadPool != null ? new Server(threadPool) : new Server();
    }
1119526
private ExecutorService _createExecutorService() {
        int producerConnections = Math.max(Integer.valueOf(
            _configuration.getValue(Property.KAFKA_PRODUCER_CONNECTIONS.getName(), Property.KAFKA_PRODUCER_CONNECTIONS.getDefaultValue())), 10);

        return Executors.newFixedThreadPool(producerConnections, new ThreadFactory() {

                AtomicInteger id = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, MessageFormat.format("kafka-producer-worker-{0}", id.getAndIncrement()));
                }
            });
    }
795261
public static ThreadFactory buildThreadFactory(@NotNull String threadNamePrefix, @NotNull boolean daemon) {
		return new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "-%d").setDaemon(daemon).build();
	}
315742
def pool(self):
        
        if self._pool is None:
            self._pool = ThreadPool(self.pool_threads)
        return self._pool
502163
def create(self):
        
        self.queue = self.scheduler.queue.addSubQueue(self.priority, LockEvent.createMatcher(self.context, self.key),
                                         maxdefault = self.size, defaultQueueClass = CBQueue.AutoClassQueue.initHelper('locker', subqueuelimit = 1))
28168
def run_in_thread_pool(self, pool_size=None, func=None, *args):
        """"""
        executor = Pool(pool_size)
        return self.loop.run_in_executor(executor, func, *args)
808872
private void startConsumerThread() {
        stopped = false;
        thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!stopped) {
                    processQueue();
                }
            }
        }, "JdbcMetricsConsumer"); //$NON-NLS-1$
        thread.setDaemon(true);
        thread.start();        
    }
1232972
private void addToPool(SessionProtocol actualProtocol, PoolKey key, PooledChannel pooledChannel) {
        assert eventLoop.inEventLoop() : Thread.currentThread().getName();
        getOrCreatePool(actualProtocol, key).addLast(pooledChannel);
    }
1328728
public static ExecutorService newPooledExecutor(String poolName, boolean useDaemonThreads)
    {
        return ParallelIterate.newPooledExecutor(ParallelIterate.getDefaultMaxThreadPoolSize(), poolName, useDaemonThreads);
    }