Intializes the partition checkpoint and lease store and then calls run async.
558392
async def start_async(self):
        
        if self.run_task:
            raise Exception("A PartitionManager cannot be started multiple times.")

        partition_count = await self.initialize_stores_async()
        _logger.info("%r PartitionCount: %r", self.host.guid, partition_count)
        self.run_task = asyncio.ensure_future(self.run_async())
600933
public void start() {
		checkState(!closed, "The RPC connection is already closed");
		checkState(!isConnected() && pendingRegistration == null, "The RPC connection is already started");

		final RetryingRegistration<F, G, S> newRegistration = createNewRegistration();

		if (REGISTRATION_UPDATER.compareAndSet(this, null, newRegistration)) {
			newRegistration.startRegistration();
		} else {
			// concurrent start operation
			newRegistration.cancel();
		}
	}
1025952
public void run() {
		ConnectionHandle connection = null;
		long tmp;
		long nextCheckInMs = this.maxAgeInMs;

		int partitionSize= this.partition.getAvailableConnections();
		long currentTime = System.currentTimeMillis();
		for (int i=0; i < partitionSize; i++){
			try {
				connection = this.partition.getFreeConnections().poll();

				if (connection != null){
					connection.setOriginatingPartition(this.partition);

					tmp = this.maxAgeInMs - (currentTime - connection.getConnectionCreationTimeInMs()); 

					if (tmp < nextCheckInMs){
						nextCheckInMs = tmp; 
					}

					if (connection.isExpired(currentTime)){
						// kill off this connection
						closeConnection(connection);
						continue;
					}


					
					if (this.lifoMode){
						// we can't put it back normally or it will end up in front again.
						if (!(connection.getOriginatingPartition().getFreeConnections().offer(connection))){
							connection.internalClose();
						}
					} else {
						this.pool.putConnectionBackInPartition(connection);
					}


					Thread.sleep(20L); // test slowly, this is not an operation that we're in a hurry to deal with (avoid CPU spikes)...
				}
			}  catch (Throwable e) {
					logger.error("Connection max age thread exception.", e);
			}

		} // throw it back on the queue

	}
953096
final static void init() {
		StopWatch stopWatch = new StopWatch();
		stopWatch.start("Audit4jInit");
		if (configContext == null) {
			configContext = new ConcurrentConfigurationContext();
		}

		if (lifeCycle.getStatus().equals(RunStatus.READY) || lifeCycle.getStatus().equals(RunStatus.STOPPED)) {
			Audit4jBanner banner = new Audit4jBanner();
			banner.printBanner();
			Log.info("Initializing Audit4j...");
			
			// Check system environment;
			checkEnvironment();
			
			// Load configurations to Memory
			Log.info("Loading Configurations...");
			if (conf == null) {
				loadConfig();
			}
			Log.info("Validating Configurations...");
			if (conf == null) {
				terminate();
				throw new InitializationException(INIT_FAILED);
			}
			try {
				ValidationManager.validateConfigurations(conf);
			} catch (ValidationException e1) {
				terminate();
				throw new InitializationException(INIT_FAILED, e1);
			}

			// Execute commands.
			CommandProcessor.getInstance().process(conf.getCommands());

			// Load Registry configurations.
			loadRegistry();

			if (conf.getProperties() == null) {
				conf.setProperties(new HashMap<String, String>());
			} else {
				for (Map.Entry<String, String> entry : conf.getProperties().entrySet()) {
					if (System.getProperties().containsKey(entry.getValue())) {
						conf.getProperties().put(entry.getKey(), System.getProperty(entry.getValue()));
					}
				}
			}
			configContext.getProperties().putAll(conf.getProperties());

			// Initialize handlers.
			initHandlers();

			// Initialize layouts.
			initLayout();

			// Initialize annotation transformer.

			if (conf.getAnnotationTransformer() == null) {
				annotationTransformer = getDefaultAnnotationTransformer();
			} else {
				annotationTransformer = conf.getAnnotationTransformer();
			}

			// Initialize IO streams.
			initStreams();

			if (!conf.getFilters().isEmpty()) {
			    Log.info("Registoring Filters...");
            }
			for (AuditEventFilter filter : conf.getFilters()) {
				configContext.addFilter(filter);
				Log.info(filter.getClass().getName() +  " Registored...");
			}

			configContext.setMetaData(conf.getMetaData());

			// Execute Scheduler tasks.
			Log.info("Executing Schedulers...");
			Schedulers.taskRegistry().scheduleAll();

			// Initialize monitoring support if available in the configurations.
			if (conf.getJmx() != null) {
				MBeanAgent agent = new MBeanAgent();
				agent.setJmxConfig(conf.getJmx());
				agent.init();
				agent.registerMbeans();
			}

			lifeCycle.setStatus(RunStatus.RUNNING);
			lifeCycle.setStartUpTime(new Date().getTime());
			stopWatch.stop();
			Long initializationTime = stopWatch.getLastTaskTimeMillis();
			Log.info("Audit4j initialized. Total time: ", initializationTime, "ms");
		}
	}
945214
public void startOnInit(Result<I> result)
  {
    
    DeployFactory2<I> builder = builder();
    
    if (builder == null) {
      result.ok(null);
      return;
    }
    
    if (! _lifecycle.toInit()) {
      result.ok(get());
      return;
    }
    
    _strategy.startOnInit(this, result);
  }
744073
private void init() {
        assertUnlocked();
        
        lock.lock();
        try {
            httpSender.init(cfg);

            LOG.info(
                "Starting with "
                    + DEFAULT_REQ_PROC_COUNT + " request processors");

            procThreads = new RequestProcessor[DEFAULT_REQ_PROC_COUNT];

            for (int i = 0; i < procThreads.length; i++) {
                procThreads[i] = new RequestProcessor(i);
                procThreads[i].start();
            }

        } finally {
            lock.unlock();
        }
    }
534946
@FFDCIgnore(JobStoppingException.class)
    private void executeAndWaitForCompletion(PartitionPlanDescriptor currentPlan) throws JobRestartException {
        if (isStoppingStoppedOrFailed()) {
            logger.fine("Job already in "
                        + runtimeWorkUnitExecution.getWorkUnitJobContext().getBatchStatus().toString()
                        + " state, exiting from executeAndWaitForCompletion() before beginning execution");
            return;
        }

        List<Integer> partitionsToExecute = getPartitionNumbersToExecute(currentPlan);

        logger.fine("Partitions to execute in this run: " + partitionsToExecute
                    + ".  Total number of partitions in step: " + currentPlan.getNumPartitionsInPlan());

        List<Integer> startedPartitions = new ArrayList<Integer>();
        List<Integer> finishedPartitions = new ArrayList<Integer>();
        List<Throwable> analyzerExceptions = new ArrayList<Throwable>();

        // Keep looping until all partitions have finished.
        while (finishedPartitions.size() < partitionsToExecute.size()) {

            startPartitions(partitionsToExecute, startedPartitions, finishedPartitions, currentPlan);

            // Check that there are still un-finished partitions running.
            // If not, break out of the loop.
            if (finishedPartitions.size() >= partitionsToExecute.size()) {
                break;
            }

            //Break this loop when nextPartitionFinished is -1
            try {
                waitForNextPartitionToFinish(analyzerExceptions, finishedPartitions);
            } catch (JobStoppingException e) {
                // break the loop
                break;
            }
        }

        if (!analyzerExceptions.isEmpty()) {
            rollbackPartitionedStep();
            throw new BatchContainerRuntimeException("Exception previously thrown by Analyzer, rolling back step.", analyzerExceptions.get(0));
        }
    }
290907
private void init() {
        logger.info("metadata init().");

        writeLock.lock();
        try {
        // Required keys
        initCache(CLUSTER_KEY);

        // If stores definition storage engine is not null, initialize metadata
        // Add the mapping from key to the storage engine used
        if(this.storeDefinitionsStorageEngine != null) {
            initStoreDefinitions(null);
        } else {
            initCache(STORES_KEY);
        }

        // Initialize system store in the metadata cache
        initSystemCache();
        initSystemRoutingStrategies(getCluster());

        // Initialize with default if not present
        initCache(SLOP_STREAMING_ENABLED_KEY, true);
        initCache(PARTITION_STREAMING_ENABLED_KEY, true);
        initCache(READONLY_FETCH_ENABLED_KEY, true);
        initCache(QUOTA_ENFORCEMENT_ENABLED_KEY, true);
        initCache(REBALANCING_STEAL_INFO, new RebalancerState(new ArrayList<RebalanceTaskInfo>()));
        initCache(SERVER_STATE_KEY, VoldemortState.NORMAL_SERVER.toString());
        initCache(REBALANCING_SOURCE_CLUSTER_XML, null);
        initCache(REBALANCING_SOURCE_STORES_XML, null);


        } finally {
            writeLock.unlock();
        }
    }
550444
def run_checkpointing
      # task check pointer is scoped to separate class to avoid
      # the secondary thread and main thread modifying the same @task
      # variable (and accidentally clobbering it in the process)
      task_checkpointer = TaskCheckPointer.new(@task.id)
      Thread.new do
        with_thread_name("task:#{@task.id}-checkpoint") do
          while true
            sleep(Config.task_checkpoint_interval)
            task_checkpointer.checkpoint
          end
        end
      end
    end
988069
public void initEvictor() {
        MapEvictionPolicy mapEvictionPolicy = getMapEvictionPolicy();
        if (mapEvictionPolicy == null) {
            evictor = NULL_EVICTOR;
        } else {
            MemoryInfoAccessor memoryInfoAccessor = getMemoryInfoAccessor();
            EvictionChecker evictionChecker = new EvictionChecker(memoryInfoAccessor, mapServiceContext);
            NodeEngine nodeEngine = mapServiceContext.getNodeEngine();
            IPartitionService partitionService = nodeEngine.getPartitionService();
            int batchSize = nodeEngine.getProperties().getInteger(MAP_EVICTION_BATCH_SIZE);
            evictor = new EvictorImpl(mapEvictionPolicy, evictionChecker, partitionService, batchSize);
        }
    }
156762
public void checkpointStart()
  {
    if (_isWhileCheckpoint) {
      throw new IllegalStateException();
    }
    
    _isWhileCheckpoint = true;
    
    long sequence = ++_sequence;
    
    setSequence(sequence);
    
    byte []headerBuffer = _headerBuffer;
    BitsUtil.writeInt16(headerBuffer, 0, CHECKPOINT_START);
    writeImpl(headerBuffer, 0, 2);
    
    _lastCheckpointStart = _index - _startAddress;
  }
243739
@Override
    public void run() {
        try {
            VoltTable partitionKeys = null;
            partitionKeys = m_client.callProcedure("@GetPartitionKeys", "INTEGER").getResults()[0];

            while (partitionKeys.advanceRow()) {
                m_client.callProcedure(new NullCallback(), "DeleteOldAdRequests",
                        partitionKeys.getLong("PARTITION_KEY"),
                        m_expiredAgeInSeconds);
            }

            m_client.callProcedure(new NullCallback(), "DeleteExpiredBids");
        }
        catch (IOException | ProcCallException ex) {
            ex.printStackTrace();
        }
    }
1246148
public static synchronized void initialize(
            int myHostId,
            CatalogContext catalogContext,
            boolean isRejoin,
            boolean forceCreate,
            HostMessenger messenger,
            List<Pair<Integer, Integer>> partitions)
            throws ExportManager.SetupException
    {
        ExportManager em = new ExportManager(myHostId, catalogContext, messenger);
        m_self = em;
        if (forceCreate) {
            em.clearOverflowData();
        }
        em.initialize(catalogContext, partitions, isRejoin);

        RealVoltDB db=(RealVoltDB)VoltDB.instance();
        db.getStatsAgent().registerStatsSource(StatsSelector.EXPORT,
                myHostId, // m_siteId,
                em.getExportStats());
    }
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();
        }
    }
934549
CompletableFuture<Partition> connect(
      PartitionManagementService managementService,
      ThreadContextFactory threadFactory) {
    election = managementService.getElectionService().getElectionFor(partitionId);
    client = new PrimaryBackupPartitionClient(this, managementService, threadFactory);
    return client.start().thenApply(v -> this);
  }
30080
def run(self):
    """"""
    while not self._state.is_terminating:
      self._maybe_garbage_collect()
      self._maybe_extend_lease()
      # Waiting with a timeout in maybe_pause has the effect of waiting until:
      # 1) we are paused and then resumed
      # 2) we are terminated (which will break the loop)
      # 3) the timeout is reached, which will cause us to wake up and check gc/leases
      self._state.maybe_pause(timeout=10)
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);
				}
			}
		}
	}