Spawns a kafka connection thread
196386
def _spawn_kafka_connection_thread(self):
        """"""
        self.logger.debug("Spawn kafka connection thread")
        self.kafka_connected = False
        self._kafka_thread = Thread(target=self._setup_kafka)
        self._kafka_thread.setDaemon(True)
        self._kafka_thread.start()
513605
def start(self):
        '''
        
        '''
        log.debug('Creating the consumer using the bootstrap servers: %s and the group ID: %s',
                  self.bootstrap_servers,
                  self.group_id)
        try:
            self.consumer = kafka.KafkaConsumer(bootstrap_servers=self.bootstrap_servers,
                                                group_id=self.group_id)
        except kafka.errors.NoBrokersAvailable as err:
            log.error(err, exc_info=True)
            raise ListenerException(err)
        log.debug('Subscribing to the %s topic', self.topic)
        self.consumer.subscribe(topics=[self.topic])
591273
def gevent_spawn(self):
        
        monkey.patch_all(thread=False)
        joinall([spawn(self.gevent_worker) for x in range(self.queue_worker_amount)])
875561
public void startup() throws InterruptedException {
        final int maxCacheConnectionPerThread = serverConfig.getMaxConnections() / processors.length;
        logger.debug("start {} Processor threads",processors.length);
        for (int i = 0; i < processors.length; i++) {
            processors[i] = new Processor(handlerFactory, stats, maxRequestSize, maxCacheConnectionPerThread);
            Utils.newThread("jafka-processor-" + i, processors[i], false).start();
        }
        Utils.newThread("jafka-acceptor", acceptor, false).start();
        acceptor.awaitStartup();
    }
1182000
def _setup_kafka(self):
        
        # close older connections
        if self.consumer is not None:
            self.logger.debug("Closing existing kafka consumer")
            self.consumer.close()
            self.consumer = None
        if self.producer is not None:
            self.logger.debug("Closing existing kafka producer")
            self.producer.flush()
            self.producer.close(timeout=10)
            self.producer = None

        # create new connections
        self._consumer_thread = None
        self.logger.debug("Creating kafka connections")
        self.consumer = self._create_consumer()
        if not self.closed:
            self.logger.debug("Kafka Conumer created")
        self.producer = self._create_producer()
        if not self.closed:
            self.logger.debug("Kafka Producer created")

        if not self.closed:
            self.kafka_connected = True
            self.logger.info("Connected successfully to Kafka")
            self._spawn_kafka_consumer_thread()
636242
def connect_async(self, conn_id, connection_string, callback):
        """"""

        future = self._loop.launch_coroutine(self._adapter.connect(conn_id, connection_string))
        future.add_done_callback(lambda x: self._callback_future(conn_id, x, callback))
171079
def run():
    """"""
    args = parse_args()
    logging.basicConfig(level=logging.WARN)

    # to prevent flooding for sensu-check.
    logging.getLogger('kafka').setLevel(logging.CRITICAL)

    if args.controller_only and args.first_broker_only:
        terminate(
            status_code.WARNING,
            prepare_terminate_message(
                "Only one of controller_only and first_broker_only should be used",
            ),
            args.json,
        )

    if args.controller_only or args.first_broker_only:
        if args.broker_id is None:
            terminate(
                status_code.WARNING,
                prepare_terminate_message("broker_id is not specified"),
                args.json,
            )
        elif args.broker_id == -1:
            try:
                args.broker_id = get_broker_id(args.data_path)
            except Exception as e:
                terminate(
                    status_code.WARNING,
                    prepare_terminate_message("{}".format(e)),
                    args.json,
                )

    try:
        cluster_config = config.get_cluster_config(
            args.cluster_type,
            args.cluster_name,
            args.discovery_base_path,
        )
        code, msg = args.command(cluster_config, args)
    except ConfigurationError as e:
        terminate(
            status_code.CRITICAL,
            prepare_terminate_message("ConfigurationError {0}".format(e)),
            args.json,
        )

    terminate(code, msg, args.json)
203445
private function processConnectionPool(): void
    {
        $readableConnections = $this->connectionPool->getReadableConnections(5);

        foreach ($readableConnections as $id => $connection) {
            if (!isset($this->connectionHandlers[$id])) {
                $this->connectionHandlers[$id] = $this->connectionHandlerFactory->createConnectionHandler($this->kernel, $connection);
            }

            try {
                $statusCodes = $this->connectionHandlers[$id]->ready();
                $this->considerStatusCodes($statusCodes);
            } catch (UserlandDaemonException $exception) {
                $this->daemonOptions->getOption(DaemonOptions::LOGGER)->error($exception->getMessage());
            }

            if ($this->connectionHandlers[$id]->isClosed()) {
                unset($this->connectionHandlers[$id]);
            }
        }
    }
494157
def run(self):
        """"""
        log.debug("Starting Kafka producer I/O thread.")

        # main loop, runs until close is called
        while self._running:
            try:
                self.run_once()
            except Exception:
                log.exception("Uncaught error in kafka producer I/O thread")

        log.debug("Beginning shutdown of Kafka producer I/O thread, sending"
                  " remaining records.")

        # okay we stopped accepting requests but there may still be
        # requests in the accumulator or waiting for acknowledgment,
        # wait until these are completed.
        while (not self._force_close
               and (self._accumulator.has_unsent()
                    or self._client.in_flight_request_count() > 0)):
            try:
                self.run_once()
            except Exception:
                log.exception("Uncaught error in kafka producer I/O thread")

        if self._force_close:
            # We need to fail all the incomplete batches and wake up the
            # threads waiting on the futures.
            self._accumulator.abort_incomplete_batches()

        try:
            self._client.close()
        except Exception:
            log.exception("Failed to close network client")

        log.debug("Shutdown of Kafka producer I/O thread has completed.")
1259995
def _send_to_kafka(self, master):
        '''
        
        '''
        appid_topic = "{prefix}.outbound_{appid}".format(
                                                    prefix=self.topic_prefix,
                                                    appid=master['appid'])
        firehose_topic = "{prefix}.outbound_firehose".format(
                                                    prefix=self.topic_prefix)
        try:
            # dont want logger in outbound kafka message
            if self.use_appid_topics:
                f1 = self.producer.send(appid_topic, master)
                f1.add_callback(self._kafka_success)
                f1.add_errback(self._kafka_failure)
            f2 = self.producer.send(firehose_topic, master)
            f2.add_callback(self._kafka_success)
            f2.add_errback(self._kafka_failure)

            return True
        except Exception as ex:
            message = "An exception '{0}' occured while sending a message " \
                "to kafka. Arguments:\n{1!r}" \
                .format(type(ex).__name__, ex.args)
            self.logger.error(message)

        return False
410403
@Override
	public void run() {
		LOG.info("Starting to fetch from {}", this.partitions);

		// set up the config values
		try {
			// create the Kafka consumer that we actually use for fetching
			consumer = new SimpleConsumer(broker.host(), broker.port(), soTimeout, bufferSize, clientId);

			// replace earliest of latest starting offsets with actual offset values fetched from Kafka
			requestAndSetEarliestOrLatestOffsetsFromKafka(consumer, partitions);

			LOG.info("Starting to consume {} partitions with consumer thread {}", partitions.size(), getName());

			// Now, the actual work starts :-)
			int offsetOutOfRangeCount = 0;
			int reconnects = 0;
			while (running) {

				// ----------------------------------- partitions list maintenance ----------------------------

				// check queue for new partitions to read from:
				List<KafkaTopicPartitionState<TopicAndPartition>> newPartitions = newPartitionsQueue.pollBatch();
				if (newPartitions != null) {
					// found some new partitions for this thread's broker

					// the new partitions should already be assigned a starting offset
					checkAllPartitionsHaveDefinedStartingOffsets(newPartitions);
					// if the new partitions are to start from earliest or latest offsets,
					// we need to replace them with actual values from Kafka
					requestAndSetEarliestOrLatestOffsetsFromKafka(consumer, newPartitions);

					// add the new partitions (and check they are not already in there)
					for (KafkaTopicPartitionState<TopicAndPartition> newPartition: newPartitions) {
						if (partitions.contains(newPartition)) {
							throw new IllegalStateException("Adding partition " + newPartition +
									" to subscribed partitions even though it is already subscribed");
						}
						partitions.add(newPartition);
					}

					LOG.info("Adding {} new partitions to consumer thread {}", newPartitions.size(), getName());
					LOG.debug("Partitions list: {}", newPartitions);
				}

				if (partitions.size() == 0) {
					if (newPartitionsQueue.close()) {
						// close succeeded. Closing thread
						running = false;

						LOG.info("Consumer thread {} does not have any partitions assigned anymore. Stopping thread.",
								getName());

						// add the wake-up marker into the queue to make the main thread
						// immediately wake up and termination faster
						unassignedPartitions.add(MARKER);

						break;
					} else {
						// close failed: fetcher main thread concurrently added new partitions into the queue.
						// go to top of loop again and get the new partitions
						continue;
					}
				}

				// ----------------------------------- request / response with kafka ----------------------------

				FetchRequestBuilder frb = new FetchRequestBuilder();
				frb.clientId(clientId);
				frb.maxWait(maxWait);
				frb.minBytes(minBytes);

				for (KafkaTopicPartitionState<?> partition : partitions) {
					frb.addFetch(
							partition.getKafkaTopicPartition().getTopic(),
							partition.getKafkaTopicPartition().getPartition(),
							partition.getOffset() + 1, // request the next record
							fetchSize);
				}

				kafka.api.FetchRequest fetchRequest = frb.build();
				LOG.debug("Issuing fetch request {}", fetchRequest);

				FetchResponse fetchResponse;
				try {
					fetchResponse = consumer.fetch(fetchRequest);
				}
				catch (Throwable cce) {
					//noinspection ConstantConditions
					if (cce instanceof ClosedChannelException) {
						LOG.warn("Fetch failed because of ClosedChannelException.");
						LOG.debug("Full exception", cce);

						// we don't know if the broker is overloaded or unavailable.
						// retry a few times, then return ALL partitions for new leader lookup
						if (++reconnects >= reconnectLimit) {
							LOG.warn("Unable to reach broker after {} retries. Returning all current partitions", reconnectLimit);
							for (KafkaTopicPartitionState<TopicAndPartition> fp: this.partitions) {
								unassignedPartitions.add(fp);
							}
							this.partitions.clear();
							continue; // jump to top of loop: will close thread or subscribe to new partitions
						}
						try {
							consumer.close();
						} catch (Throwable t) {
							LOG.warn("Error while closing consumer connection", t);
						}
						// delay & retry
						Thread.sleep(100);
						consumer = new SimpleConsumer(broker.host(), broker.port(), soTimeout, bufferSize, clientId);
						continue; // retry
					} else {
						throw cce;
					}
				}
				reconnects = 0;

				// ---------------------------------------- error handling ----------------------------

				if (fetchResponse == null) {
					throw new IOException("Fetch from Kafka failed (request returned null)");
				}

				if (fetchResponse.hasError()) {
					String exception = "";
					List<KafkaTopicPartitionState<TopicAndPartition>> partitionsToGetOffsetsFor = new ArrayList<>();

					// iterate over partitions to get individual error codes
					Iterator<KafkaTopicPartitionState<TopicAndPartition>> partitionsIterator = partitions.iterator();
					boolean partitionsRemoved = false;

					while (partitionsIterator.hasNext()) {
						final KafkaTopicPartitionState<TopicAndPartition> fp = partitionsIterator.next();
						short code = fetchResponse.errorCode(fp.getTopic(), fp.getPartition());

						if (code == ErrorMapping.OffsetOutOfRangeCode()) {
							// we were asked to read from an out-of-range-offset (maybe set wrong in Zookeeper)
							// Kafka's high level consumer is resetting the offset according to 'auto.offset.reset'
							partitionsToGetOffsetsFor.add(fp);
						}
						else if (code == ErrorMapping.NotLeaderForPartitionCode() ||
								code == ErrorMapping.LeaderNotAvailableCode() ||
								code == ErrorMapping.BrokerNotAvailableCode() ||
								code == ErrorMapping.UnknownCode()) {
							// the broker we are connected to is not the leader for the partition.
							LOG.warn("{} is not the leader of {}. Reassigning leader for partition", broker, fp);
							LOG.debug("Error code = {}", code);

							unassignedPartitions.add(fp);

							partitionsIterator.remove(); // unsubscribe the partition ourselves
							partitionsRemoved = true;
						}
						else if (code != ErrorMapping.NoError()) {
							exception += "\nException for " + fp.getTopic() + ":" + fp.getPartition() + ": " +
									ExceptionUtils.stringifyException(ErrorMapping.exceptionFor(code));
						}
					}
					if (partitionsToGetOffsetsFor.size() > 0) {
						// safeguard against an infinite loop.
						if (offsetOutOfRangeCount++ > 3) {
							throw new RuntimeException("Found invalid offsets more than three times in partitions "
									+ partitionsToGetOffsetsFor + " Exceptions: " + exception);
						}
						// get valid offsets for these partitions and try again.
						LOG.warn("The following partitions had an invalid offset: {}", partitionsToGetOffsetsFor);
						requestAndSetSpecificTimeOffsetsFromKafka(consumer, partitionsToGetOffsetsFor, invalidOffsetBehavior);

						LOG.warn("The new partition offsets are {}", partitionsToGetOffsetsFor);
						continue; // jump back to create a new fetch request. The offset has not been touched.
					}
					else if (partitionsRemoved) {
						continue; // create new fetch request
					}
					else {
						// partitions failed on an error
						throw new IOException("Error while fetching from broker '" + broker + "': " + exception);
					}
				} else {
					// successful fetch, reset offsetOutOfRangeCount.
					offsetOutOfRangeCount = 0;
				}

				// ----------------------------------- process fetch response ----------------------------

				int messagesInFetch = 0;
				int deletedMessages = 0;
				Iterator<KafkaTopicPartitionState<TopicAndPartition>> partitionsIterator = partitions.iterator();

				partitionsLoop:
				while (partitionsIterator.hasNext()) {
					final KafkaTopicPartitionState<TopicAndPartition> currentPartition = partitionsIterator.next();

					final ByteBufferMessageSet messageSet = fetchResponse.messageSet(
							currentPartition.getTopic(), currentPartition.getPartition());

					for (MessageAndOffset msg : messageSet) {
						if (running) {
							messagesInFetch++;
							final ByteBuffer payload = msg.message().payload();
							final long offset = msg.offset();

							if (offset <= currentPartition.getOffset()) {
								// we have seen this message already
								LOG.info("Skipping message with offset " + msg.offset()
										+ " because we have seen messages until (including) "
										+ currentPartition.getOffset()
										+ " from topic/partition " + currentPartition.getTopic() + '/'
										+ currentPartition.getPartition() + " already");
								continue;
							}

							// If the message value is null, this represents a delete command for the message key.
							// Log this and pass it on to the client who might want to also receive delete messages.
							byte[] valueBytes;
							if (payload == null) {
								deletedMessages++;
								valueBytes = null;
							} else {
								valueBytes = new byte[payload.remaining()];
								payload.get(valueBytes);
							}

							// put key into byte array
							byte[] keyBytes = null;
							int keySize = msg.message().keySize();

							if (keySize >= 0) { // message().hasKey() is doing the same. We save one int deserialization
								ByteBuffer keyPayload = msg.message().key();
								keyBytes = new byte[keySize];
								keyPayload.get(keyBytes);
							}

							final T value = deserializer.deserialize(
								new ConsumerRecord<>(
									currentPartition.getTopic(),
									currentPartition.getPartition(), keyBytes, valueBytes, offset));

							if (deserializer.isEndOfStream(value)) {
								// remove partition from subscribed partitions.
								partitionsIterator.remove();
								continue partitionsLoop;
							}

							owner.emitRecord(value, currentPartition, offset);
						}
						else {
							// no longer running
							return;
						}
					}
				}
				LOG.debug("This fetch contained {} messages ({} deleted messages)", messagesInFetch, deletedMessages);
			} // end of fetch loop

			if (!newPartitionsQueue.close()) {
				throw new Exception("Bug: Cleanly leaving fetcher thread without having a closed queue.");
			}
		}
		catch (Throwable t) {
			// report to the fetcher's error handler
			errorHandler.reportError(t);
		}
		finally {
			if (consumer != null) {
				// closing the consumer should not fail the program
				try {
					consumer.close();
				}
				catch (Throwable t) {
					LOG.error("Error while closing the Kafka simple consumer", t);
				}
			}
		}
	}
1199686
private SimpleConsumerThread<T> createAndStartSimpleConsumerThread(
			List<KafkaTopicPartitionState<TopicAndPartition>> seedPartitions,
			Node leader,
			ExceptionProxy errorHandler) throws IOException, ClassNotFoundException {
		// each thread needs its own copy of the deserializer, because the deserializer is
		// not necessarily thread safe
		final KafkaDeserializationSchema<T> clonedDeserializer =
				InstantiationUtil.clone(deserializer, runtimeContext.getUserCodeClassLoader());

		// seed thread with list of fetch partitions (otherwise it would shut down immediately again
		SimpleConsumerThread<T> brokerThread = new SimpleConsumerThread<>(
				this, errorHandler, kafkaConfig, leader, seedPartitions, unassignedPartitionsQueue,
				clonedDeserializer, invalidOffsetBehavior);

		brokerThread.setName(String.format("SimpleConsumer - %s - broker-%s (%s:%d)",
				runtimeContext.getTaskName(), leader.id(), leader.host(), leader.port()));
		brokerThread.setDaemon(true);
		brokerThread.start();

		LOG.info("Starting thread {}", brokerThread.getName());
		return brokerThread;
	}
1036399
def run(connection)
      @connection = connection
      bot = create_bot(connection)
      thread = Thread.new { bot.start }
      @ready.pop
      loop { deliver bot, dequeue }
    ensure
      thread.exit unless thread.nil?
    end
1184239
public void threadStarted(ThreadPool tp, int activeThreads, int maxThreads)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "threadStarted", new Object[]{tp, new Integer(activeThreads), new Integer(maxThreads)});
    
    if (TraceComponent.isAnyTracingEnabled() && (tc.isDebugEnabled() || tc.isDumpEnabled() || tc.isEntryEnabled() || tc.isEventEnabled()))
      SibTr.push(_messageProcessor.getMessagingEngine());  
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.exit(tc, "threadStarted");
  }
133947
protected void processConnection(WatchedEvent event) {
    switch (event.getState()) {
    case SyncConnected:
      LOG.info(hostPort.getHostPort() + " sync connect from ZooKeeper");
      try {
        waitToInitZooKeeper(2000); // init zookeeper in another thread, wait for a while
      } catch (Exception e) {
        LOG.fatal("Error to init ZooKeeper object after sleeping 2000 ms, exit immediately");
        System.exit(0);
      } 
      break;
    case Disconnected: // be triggered when kill the server or the leader of zk cluster 
      LOG.warn(hostPort.getHostPort() + " received disconnected from ZooKeeper");
      break;
    case AuthFailed:
      LOG.fatal(hostPort.getHostPort() + " auth fail, exit immediately");
      System.exit(0);
    case Expired:
      LOG.fatal(hostPort.getHostPort() + " received expired from ZooKeeper, exit immediately");
      System.exit(0);
      break;
    default:
      break;
    }
  }
98173
protected void startReclaimConnectionThread() {

        if (tc.isEntryEnabled()) {
            Tr.entry(this, tc, "startReclaimConnectionThread");
        }

        //long reapTime;

        synchronized (taskTimerLockObject) {
            if (!reaperThreadStarted) {
                if (reapTime > 0) {
                    if ((this.unusedTimeout > 0)
                        || (this.agedTimeout > 0)) {
                        final PoolManager tempPM = this;
                        reaperThreadStarted = true;
                        AccessController.doPrivileged(new PrivilegedAction<Void>() {
                            @Override
                            public Void run() {
                                new TaskTimer(tempPM);
                                return null;
                            }
                        });

                        if (tc.isDebugEnabled()) {
                            Tr.debug(this, tc, "Started reclaim connection thread for pool " + gConfigProps.getXpathId());
                        }
                    }

                }
            }
        }

        if (tc.isEntryEnabled()) {
            Tr.exit(this, tc, "startReclaimConnectionThread");
        }

    }
152616
def spawn_process(self, target, *args):
        
        p = Process(target=target, args=args)
        p.daemon = True
        if target == worker:
            p.daemon = Conf.DAEMONIZE_WORKERS
            p.timer = args[2]
            self.pool.append(p)
        p.start()
        return p