This method also starts the collector thread lazily.
548458
@Override
    public void log(StopWatch sw)
    {
        if( !m_queue.offer( sw.freeze() ) ) m_rejectedStopWatches.getAndIncrement();

        if( m_collectorThread == null )
        {
            synchronized(this)
            {
                //
                //  Ensure that there is no race condition starting the thread.
                //  Note that under the JVM5 memory model this also requires
                //  that the field is declared as "volatile", or else the compiler
                //  may do something nasty.
                //
                if( m_collectorThread == null )
                {
                    m_collectorThread = new CollectorThread();
                    m_collectorThread.setName( "Speed4J PeriodicalLog Collector Thread" );
                    m_collectorThread.setDaemon( true );
                    m_collectorThread.start();
                }
            }

        }
    }
462527
public void start() throws Throwable
   {
      if (!isExternal)
      {
         executorService = Executors.newSingleThreadExecutor(new ValidatorThreadFactory());
      }

      shutdown.set(false);
      interval = Long.MAX_VALUE;
      next = Long.MAX_VALUE;

      executorService.execute(new ConnectionValidatorRunner());
   }
410286
@Override
	public void runFetchLoop() throws Exception {
		// the map from broker to the thread that is connected to that broker
		final Map<Node, SimpleConsumerThread<T>> brokerToThread = new HashMap<>();

		// this holds possible the exceptions from the concurrent broker connection threads
		final ExceptionProxy errorHandler = new ExceptionProxy(Thread.currentThread());

		// the offset handler handles the communication with ZooKeeper, to commit externally visible offsets
		final ZookeeperOffsetHandler zookeeperOffsetHandler = new ZookeeperOffsetHandler(kafkaConfig);
		this.zookeeperOffsetHandler = zookeeperOffsetHandler;

		PeriodicOffsetCommitter periodicCommitter = null;
		try {

			// offsets in the state may still be placeholder sentinel values if we are starting fresh, or the
			// checkpoint / savepoint state we were restored with had not completely been replaced with actual offset
			// values yet; replace those with actual offsets, according to what the sentinel value represent.
			for (KafkaTopicPartitionState<TopicAndPartition> partition : subscribedPartitionStates()) {
				if (partition.getOffset() == KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET) {
					// this will be replaced by an actual offset in SimpleConsumerThread
					partition.setOffset(OffsetRequest.EarliestTime());
				} else if (partition.getOffset() == KafkaTopicPartitionStateSentinel.LATEST_OFFSET) {
					// this will be replaced by an actual offset in SimpleConsumerThread
					partition.setOffset(OffsetRequest.LatestTime());
				} else if (partition.getOffset() == KafkaTopicPartitionStateSentinel.GROUP_OFFSET) {
					Long committedOffset = zookeeperOffsetHandler.getCommittedOffset(partition.getKafkaTopicPartition());
					if (committedOffset != null) {
						// the committed offset in ZK represents the next record to process,
						// so we subtract it by 1 to correctly represent internal state
						partition.setOffset(committedOffset - 1);
					} else {
						// if we can't find an offset for a partition in ZK when using GROUP_OFFSETS,
						// we default to "auto.offset.reset" like the Kafka high-level consumer
						LOG.warn("No group offset can be found for partition {} in Zookeeper;" +
							" resetting starting offset to 'auto.offset.reset'", partition);

						partition.setOffset(invalidOffsetBehavior);
					}
				} else {
					// the partition already has a specific start offset and is ready to be consumed
				}
			}

			// start the periodic offset committer thread, if necessary
			if (autoCommitInterval > 0) {
				LOG.info("Starting periodic offset committer, with commit interval of {}ms", autoCommitInterval);

				periodicCommitter = new PeriodicOffsetCommitter(
						zookeeperOffsetHandler,
						subscribedPartitionStates(),
						errorHandler,
						autoCommitInterval);
				periodicCommitter.setName("Periodic Kafka partition offset committer");
				periodicCommitter.setDaemon(true);
				periodicCommitter.start();
			}

			// Main loop polling elements from the unassignedPartitions queue to the threads
			while (running) {
				// re-throw any exception from the concurrent fetcher threads
				errorHandler.checkAndThrowException();

				// wait for max 5 seconds trying to get partitions to assign
				// if threads shut down, this poll returns earlier, because the threads inject the
				// special marker into the queue
				List<KafkaTopicPartitionState<TopicAndPartition>> partitionsToAssign =
						unassignedPartitionsQueue.getBatchBlocking(5000);
				// note: if there are more markers, remove them all
				partitionsToAssign.removeIf(MARKER::equals);

				if (!partitionsToAssign.isEmpty()) {
					LOG.info("Assigning {} partitions to broker threads", partitionsToAssign.size());
					Map<Node, List<KafkaTopicPartitionState<TopicAndPartition>>> partitionsWithLeaders =
							findLeaderForPartitions(partitionsToAssign, kafkaConfig);

					// assign the partitions to the leaders (maybe start the threads)
					for (Map.Entry<Node, List<KafkaTopicPartitionState<TopicAndPartition>>> partitionsWithLeader :
							partitionsWithLeaders.entrySet()) {
						final Node leader = partitionsWithLeader.getKey();
						final List<KafkaTopicPartitionState<TopicAndPartition>> partitions = partitionsWithLeader.getValue();
						SimpleConsumerThread<T> brokerThread = brokerToThread.get(leader);

						if (!running) {
							break;
						}

						if (brokerThread == null || !brokerThread.getNewPartitionsQueue().isOpen()) {
							// start new thread
							brokerThread = createAndStartSimpleConsumerThread(partitions, leader, errorHandler);
							brokerToThread.put(leader, brokerThread);
						}
						else {
							// put elements into queue of thread
							ClosableBlockingQueue<KafkaTopicPartitionState<TopicAndPartition>> newPartitionsQueue =
									brokerThread.getNewPartitionsQueue();

							for (KafkaTopicPartitionState<TopicAndPartition> fp : partitions) {
								if (!newPartitionsQueue.addIfOpen(fp)) {
									// we were unable to add the partition to the broker's queue
									// the broker has closed in the meantime (the thread will shut down)
									// create a new thread for connecting to this broker
									List<KafkaTopicPartitionState<TopicAndPartition>> seedPartitions = new ArrayList<>();
									seedPartitions.add(fp);
									brokerThread = createAndStartSimpleConsumerThread(seedPartitions, leader, errorHandler);
									brokerToThread.put(leader, brokerThread);
									newPartitionsQueue = brokerThread.getNewPartitionsQueue(); // update queue for the subsequent partitions
								}
							}
						}
					}
				}
				else {
					// there were no partitions to assign. Check if any broker threads shut down.
					// we get into this section of the code, if either the poll timed out, or the
					// blocking poll was woken up by the marker element
					Iterator<SimpleConsumerThread<T>> bttIterator = brokerToThread.values().iterator();
					while (bttIterator.hasNext()) {
						SimpleConsumerThread<T> thread = bttIterator.next();
						if (!thread.getNewPartitionsQueue().isOpen()) {
							LOG.info("Removing stopped consumer thread {}", thread.getName());
							bttIterator.remove();
						}
					}
				}

				if (brokerToThread.size() == 0 && unassignedPartitionsQueue.isEmpty()) {
					if (unassignedPartitionsQueue.close()) {
						LOG.info("All consumer threads are finished, there are no more unassigned partitions. Stopping fetcher");
						break;
					}
					// we end up here if somebody added something to the queue in the meantime --> continue to poll queue again
				}
			}
		}
		catch (InterruptedException e) {
			// this may be thrown because an exception on one of the concurrent fetcher threads
			// woke this thread up. make sure we throw the root exception instead in that case
			errorHandler.checkAndThrowException();

			// no other root exception, throw the interrupted exception
			throw e;
		}
		finally {
			this.running = false;
			this.zookeeperOffsetHandler = null;

			// if we run a periodic committer thread, shut that down
			if (periodicCommitter != null) {
				periodicCommitter.shutdown();
			}

			// clear the interruption flag
			// this allows the joining on consumer threads (on best effort) to happen in
			// case the initial interrupt already
			Thread.interrupted();

			// make sure that in any case (completion, abort, error), all spawned threads are stopped
			try {
				int runningThreads;
				do {
					// check whether threads are alive and cancel them
					runningThreads = 0;
					Iterator<SimpleConsumerThread<T>> threads = brokerToThread.values().iterator();
					while (threads.hasNext()) {
						SimpleConsumerThread<?> t = threads.next();
						if (t.isAlive()) {
							t.cancel();
							runningThreads++;
						} else {
							threads.remove();
						}
					}

					// wait for the threads to finish, before issuing a cancel call again
					if (runningThreads > 0) {
						for (SimpleConsumerThread<?> t : brokerToThread.values()) {
							t.join(500 / runningThreads + 1);
						}
					}
				}
				while (runningThreads > 0);
			}
			catch (InterruptedException ignored) {
				// waiting for the thread shutdown apparently got interrupted
				// restore interrupted state and continue
				Thread.currentThread().interrupt();
			}
			catch (Throwable t) {
				// we catch all here to preserve the original exception
				LOG.error("Exception while shutting down consumer threads", t);
			}

			try {
				zookeeperOffsetHandler.close();
			}
			catch (Throwable t) {
				// we catch all here to preserve the original exception
				LOG.error("Exception while shutting down ZookeeperOffsetHandler", t);
			}
		}
	}
902841
def run(self):
        
        self.collector.start()
        if self.standalone:
            self.collector.join()
1173230
@Override
  public void run() {
    synchronized (lock) {
      if (!callables.hasNext()) {
        checkEnd();
        return;
      }

      for (int i = 0; i < parallelism && callables.hasNext(); i++) {
        setupNext(callables.next());
      }
    }

    future.whenCancelled(() -> {
      cancel = true;
      checkNext();
    });
  }
1256101
public void start() {
        if (state == State.RUNNING) return;
        final Lock lock = this.lock.writeLock();
        try {
            lock.lock();
            this.state = State.STARTING;

            // Create an executor service that we'll use to start the repositories ...
            ThreadFactory threadFactory = new NamedThreadFactory("modeshape-start-repo");
            repositoryStarterService = Executors.newCachedThreadPool(threadFactory);

            state = State.RUNNING;
        } catch (RuntimeException e) {
            state = State.NOT_RUNNING;
            throw e;
        } finally {
            lock.unlock();
        }
    }
3588
private void lazyClassInit() {
        if (lazyInitDone) return;
        synchronized (lazyInitLock) {
            if (redirect!=null) {
                throw new GroovyBugError("lazyClassInit called on a proxy ClassNode, that must not happen."+
                                         "A redirect() call is missing somewhere!");
            }   
            if (lazyInitDone) return;
            VMPluginFactory.getPlugin().configureClassNode(compileUnit,this);
            lazyInitDone = true;
        }
    }
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();        
    }
9894
@Override
    protected void initQueue(T queue, QueueSpec spec) throws Exception {
        queue.setMongoClient(defaultMongoClient).setConnectionString(defaultConnectionString)
                .setCollectionName(defaultCollectionName);

        String connectionString = spec.getField(SPEC_FIELD_CONNECTION_STRING);
        if (!StringUtils.isBlank(connectionString)) {
            queue.setConnectionString(connectionString);
        }

        String collectionName = spec.getField(SPEC_FIELD_COLLECTION_NAME);
        if (!StringUtils.isBlank(collectionName)) {
            queue.setCollectionName(collectionName);
        }

        super.initQueue(queue, spec);
    }
1088104
void start() {
    /* make sure any currently running dispatchers are stopped */
    stop();

    /* create the download dispatcher and start it. */
    for (int i = 0; i < dispatchers.length; i++) {
      DownloadDispatcher dispatcher = new DownloadDispatcher(downloadQueue, delivery, logger);
      dispatchers[i] = dispatcher;
      dispatcher.start();
    }

    logger.log("Thread pool size: " + dispatchers.length);
  }
1336610
public <R> R collect(Func1<Iterable<T>, R> collector) {
        return collector.call(this);
    }
191172
protected void starting() {
        if (mementos != null) {
            for (int i = 0; i < mementos.length; i++) {
                mementos[i].cache();
            }
        }
    }
118172
private boolean mainLoop()
        throws InterruptedException
    {
        long now = System.currentTimeMillis();
        Iterator list = m_hardStore.values().iterator();
        while( list.hasNext() )
        {
            Entry entry = (Entry) list.next();
            if( entry.m_collectTime < now )
            {
                m_weakStore.put( entry.m_connection, DUMMY );
                list.remove();
            }
        }
        if( m_hardStore.size() == 0 )
        {
            m_thread = null;    // mark to start a new thread next time.
            return true;
        }
        wait( 10000 );
        return false;
    }
1000070
public void startPurge(WakeableLooper wakeableLooper) {
    synchronized (CacheCore.class) {
      if (looper == null) {
        looper = wakeableLooper;
      }

      looper.registerTimerEvent(interval, new Runnable() {
        @Override
        public void run() {
          purge();
        }
      });
    }
  }
148449
public synchronized void stopCollecting() {
		started = false;
		if (thread != null) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				// ignore
			}
			thread = null;
		}
	}
622133
private void lazyRefresh() {
    Runnable pr = new Runnable() {
      @Override
      public void run() {
        if(pendingRefresh.compareAndSet(this, null)) {
          refresh();
        }
      }
    };
    pendingRefresh.set(pr);
    scheduleUpdate(pr);
  }
539465
@Override
    public void shutdown()
    {
        m_stopCollector = true;
        if( m_collectorThread != null ) m_collectorThread.interrupt();

        try
        {
            //
            //  Remove MBean
            //
            if( m_mbeanServer != null && m_mbeanServer.isRegistered(getJMXName()) )
                m_mbeanServer.unregisterMBean(getJMXName());
        }
        catch (Exception e)
        {
            log.error("Failed to unregister MBean", e);
        }
    }