Is not scheduled, because it is performing disjoint work from sstable compaction.
170714
public Future<?> submitIndexBuild(final SecondaryIndexBuilder builder)
    {
        Runnable runnable = new Runnable()
        {
            public void run()
            {
                metrics.beginCompaction(builder);
                try
                {
                    builder.build();
                }
                finally
                {
                    metrics.finishCompaction(builder);
                }
            }
        };
        if (executor.isShutdown())
        {
            logger.info("Compaction executor has shut down, not submitting index build");
            return null;
        }

        return executor.submit(runnable);
    }
1328693
private void scheduleTasksCleanupForWorker(final String worker, final List<String> tasksToFail)
  {
    // This method is only called from the PathChildrenCache event handler, so this may look like a race,
    // but is actually not.
    cancelWorkerCleanup(worker);

    final ListenableScheduledFuture<?> cleanupTask = cleanupExec.schedule(
        new Runnable()
        {
          @Override
          public void run()
          {
            log.info("Running scheduled cleanup for Worker[%s]", worker);
            try {
              for (String assignedTask : tasksToFail) {
                String taskPath = JOINER.join(indexerZkConfig.getTasksPath(), worker, assignedTask);
                String statusPath = JOINER.join(indexerZkConfig.getStatusPath(), worker, assignedTask);
                if (cf.checkExists().forPath(taskPath) != null) {
                  cf.delete().guaranteed().forPath(taskPath);
                }

                if (cf.checkExists().forPath(statusPath) != null) {
                  cf.delete().guaranteed().forPath(statusPath);
                }

                log.info("Failing task[%s]", assignedTask);
                RemoteTaskRunnerWorkItem taskRunnerWorkItem = runningTasks.remove(assignedTask);
                if (taskRunnerWorkItem != null) {
                  taskRunnerWorkItem.setResult(TaskStatus.failure(assignedTask));
                  TaskRunnerUtils.notifyStatusChanged(listeners, assignedTask, TaskStatus.failure(assignedTask));
                } else {
                  log.warn("RemoteTaskRunner has no knowledge of task[%s]", assignedTask);
                }
              }

              // worker is gone, remove worker task status announcements path.
              String workerStatusPath = JOINER.join(indexerZkConfig.getStatusPath(), worker);
              if (cf.checkExists().forPath(workerStatusPath) != null) {
                cf.delete().guaranteed().forPath(JOINER.join(indexerZkConfig.getStatusPath(), worker));
              }
            }
            catch (Exception e) {
              log.makeAlert("Exception while cleaning up worker[%s]", worker).emit();
              throw new RuntimeException(e);
            }
          }
        },
        config.getTaskCleanupTimeout().toStandardDuration().getMillis(),
        TimeUnit.MILLISECONDS
    );

    removedWorkerCleanups.put(worker, cleanupTask);

    // Remove this entry from removedWorkerCleanups when done, if it's actually the one in there.
    Futures.addCallback(
        cleanupTask,
        new FutureCallback<Object>()
        {
          @Override
          public void onSuccess(Object result)
          {
            removedWorkerCleanups.remove(worker, cleanupTask);
          }

          @Override
          public void onFailure(Throwable t)
          {
            removedWorkerCleanups.remove(worker, cleanupTask);
          }
        }
    );
  }
36569
private void doCleanupOne(final ColumnFamilyStore cfs, SSTableReader sstable, CleanupStrategy cleanupStrategy, Collection<Range<Token>> ranges, boolean hasIndexes) throws IOException
    {
        assert !cfs.isIndex();

        Set<SSTableReader> sstableSet = Collections.singleton(sstable);

        if (!hasIndexes && !new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(ranges))
        {
            cfs.getDataTracker().markCompactedSSTablesReplaced(sstableSet, Collections.<SSTableReader>emptyList(), OperationType.CLEANUP);
            return;
        }
        if (!needsCleanup(sstable, ranges))
        {
            logger.debug("Skipping {} for cleanup; all rows should be kept", sstable);
            return;
        }

        long start = System.nanoTime();

        long totalkeysWritten = 0;

        int expectedBloomFilterSize = Math.max(cfs.metadata.getMinIndexInterval(),
                                               (int) (SSTableReader.getApproximateKeyCount(sstableSet)));
        if (logger.isDebugEnabled())
            logger.debug("Expected bloom filter size : {}", expectedBloomFilterSize);

        logger.info("Cleaning up {}", sstable);

        File compactionFileLocation = cfs.directories.getWriteableLocationAsFile(cfs.getExpectedCompactedFileSize(sstableSet, OperationType.CLEANUP));
        if (compactionFileLocation == null)
            throw new IOException("disk full");

        ISSTableScanner scanner = cleanupStrategy.getScanner(sstable, getRateLimiter());
        CleanupInfo ci = new CleanupInfo(sstable, scanner);

        metrics.beginCompaction(ci);
        Set<SSTableReader> oldSSTable = Sets.newHashSet(sstable);
        SSTableRewriter writer = new SSTableRewriter(cfs, oldSSTable, sstable.maxDataAge, false);
        List<SSTableReader> finished;
        try (CompactionController controller = new CompactionController(cfs, sstableSet, getDefaultGcBefore(cfs)))
        {
            writer.switchWriter(createWriter(cfs, compactionFileLocation, expectedBloomFilterSize, sstable.getSSTableMetadata().repairedAt, sstable));

            while (scanner.hasNext())
            {
                if (ci.isStopRequested())
                    throw new CompactionInterruptedException(ci.getCompactionInfo());

                SSTableIdentityIterator row = (SSTableIdentityIterator) scanner.next();
                row = cleanupStrategy.cleanup(row);
                if (row == null)
                    continue;
                AbstractCompactedRow compactedRow = new LazilyCompactedRow(controller, Collections.singletonList(row));
                if (writer.append(compactedRow) != null)
                    totalkeysWritten++;
            }

            // flush to ensure we don't lose the tombstones on a restart, since they are not commitlog'd
            cfs.indexManager.flushIndexesBlocking();

            finished = writer.finish();
            cfs.getDataTracker().markCompactedSSTablesReplaced(oldSSTable, finished, OperationType.CLEANUP);
        }
        catch (Throwable e)
        {
            writer.abort();
            throw Throwables.propagate(e);
        }
        finally
        {
            scanner.close();
            metrics.finishCompaction(ci);
        }

        if (!finished.isEmpty())
        {
            String format = "Cleaned up to %s.  %,d to %,d (~%d%% of original) bytes for %,d keys.  Time: %,dms.";
            long dTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
            long startsize = sstable.onDiskLength();
            long endsize = 0;
            for (SSTableReader newSstable : finished)
                endsize += newSstable.onDiskLength();
            double ratio = (double) endsize / (double) startsize;
            logger.info(String.format(format, finished.get(0).getFilename(), startsize, endsize, (int) (ratio * 100), totalkeysWritten, dTime));
        }

    }
670068
private void scheduleBatchLoadIfNeeded(Serializable id, SharedSessionContractImplementor session) throws MappingException {
		//cannot batch fetch by unique key (property-ref associations)
		if ( StringHelper.isEmpty( delegate.getRHSUniqueKeyPropertyName() ) && id != null ) {
			EntityPersister persister = session.getFactory().getMetamodel().entityPersister( delegate.getAssociatedEntityName() );
			EntityKey entityKey = session.generateEntityKey( id, persister );
			if ( !session.getPersistenceContext().containsEntity( entityKey ) ) {
				session.getPersistenceContext().getBatchFetchQueue().addBatchLoadableEntityKey( entityKey );
			}
		}
	}
283117
public void markCompactedSSTablesReplaced(Collection<SSTableReader> oldSSTables, Collection<SSTableReader> allReplacements, OperationType compactionType)
    {
        removeSSTablesFromTracker(oldSSTables);
        releaseReferences(oldSSTables, false);
        notifySSTablesChanged(oldSSTables, allReplacements, compactionType);
        addNewSSTablesSize(allReplacements);
    }
835425
public void rescheduleJob(final JobInstance toBeRescheduled) {
        final JobIdentifier jobIdentifier = toBeRescheduled.getIdentifier();
        synchronized (mutexForStageInstance(jobIdentifier)) {
            synchronized (mutexForJob(jobIdentifier)) {
                transactionTemplate.execute(new TransactionCallbackWithoutResult() {
                    @Override
                    protected void doInTransactionWithoutResult(TransactionStatus status) {
                        LOGGER.warn("[Job Reschedule] Rescheduling and marking old job as ignored: {}", toBeRescheduled);
                        //Reloading it because we want to see the latest committed state after acquiring the mutex.
                        JobInstance oldJob = jobInstanceService.buildById(toBeRescheduled.getId());
                        if (oldJob.isCompleted() || oldJob.isRescheduled()) {
                            return;
                        }
                        JobInstance newJob = oldJob.clone();
                        oldJob.changeState(JobState.Rescheduled);
                        jobInstanceService.updateStateAndResult(oldJob);

                        //Make a new Job
                        newJob.reschedule();
                        jobInstanceService.save(oldJob.getIdentifier().getStageIdentifier(), oldJob.getStageId(), newJob);

                        //Copy the plan for the old job since we don't load job plan with jobInstance by default
                        JobPlan plan = jobInstanceDao.loadPlan(oldJob.getId());
                        jobInstanceDao.ignore(oldJob);
                        jobInstanceDao.save(newJob.getId(), plan);
                        LOGGER.info("[Job Reschedule] Scheduled new job: {}. Replacing old job: {}", newJob.getIdentifier(), oldJob.getIdentifier());
                    }
                });
            }
        }
    }
634252
private TableReference executeQueryAndWaitForCompletion()
      throws IOException, InterruptedException {
    checkState(projectId != null, "Unable to execute a query without a configured project id");
    checkState(queryConfig != null, "Unable to execute a query without a configured query");
    // Dry run query to get source table location
    Job dryRunJob = new Job()
        .setConfiguration(new JobConfiguration()
            .setQuery(queryConfig)
            .setDryRun(true));
    JobStatistics jobStats = executeWithBackOff(
        client.jobs().insert(projectId, dryRunJob),
        String.format("Error when trying to dry run query %s.",
            queryConfig.toPrettyString())).getStatistics();

    // Let BigQuery to pick default location if the query does not read any tables.
    String location = null;
    @Nullable List<TableReference> tables = jobStats.getQuery().getReferencedTables();
    if (tables != null && !tables.isEmpty()) {
      Table table = getTable(tables.get(0));
      location = table.getLocation();
    }

    // Create a temporary dataset to store results.
    // Starting dataset name with an "_" so that it is hidden.
    Random rnd = new Random(System.currentTimeMillis());
    temporaryDatasetId = "_beam_temporary_dataset_" + rnd.nextInt(1000000);
    temporaryTableId = "beam_temporary_table_" + rnd.nextInt(1000000);

    createDataset(temporaryDatasetId, location);
    Job job = new Job();
    JobConfiguration config = new JobConfiguration();
    config.setQuery(queryConfig);
    job.setConfiguration(config);

    TableReference destinationTable = new TableReference();
    destinationTable.setProjectId(projectId);
    destinationTable.setDatasetId(temporaryDatasetId);
    destinationTable.setTableId(temporaryTableId);
    queryConfig.setDestinationTable(destinationTable);
    queryConfig.setAllowLargeResults(true);

    Job queryJob = executeWithBackOff(
        client.jobs().insert(projectId, job),
        String.format("Error when trying to execute the job for query %s.",
            queryConfig.toPrettyString()));
    JobReference jobId = queryJob.getJobReference();

    while (true) {
      Job pollJob = executeWithBackOff(
          client.jobs().get(projectId, jobId.getJobId()),
          String.format("Error when trying to get status of the job for query %s.",
              queryConfig.toPrettyString()));
      JobStatus status = pollJob.getStatus();
      if (status.getState().equals("DONE")) {
        // Job is DONE, but did not necessarily succeed.
        ErrorProto error = status.getErrorResult();
        if (error == null) {
          return pollJob.getConfiguration().getQuery().getDestinationTable();
        } else {
          // There will be no temporary table to delete, so null out the reference.
          temporaryTableId = null;
          throw new IOException(String.format(
              "Executing query %s failed: %s", queryConfig.toPrettyString(), error.getMessage()));
        }
      }
      Uninterruptibles.sleepUninterruptibly(
          QUERY_COMPLETION_POLL_TIME.getMillis(), TimeUnit.MILLISECONDS);
    }
  }
832825
@Override
	public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
		int counts = rejectCounts.incrementAndGet();
		if(logger.isWarnEnabled()) {
			int t = counts % 100;
			if (t == 0)
			{
				logger.warn(new StringBuilder().append("Task[Import DB Data to Elasticsearch] rejected  ").append(counts).append(" times.").toString());
			}
		}
//
		try {
			executor.getQueue().put(r);
		} catch (InterruptedException e1) {
			throw new RejectedExecutionException(e1);
		}
	}
1152599
int computeReplicationWorkForBlocks(List<List<BlockInfo>> blocksToReplicate) {
    int requiredReplication, numEffectiveReplicas, priority;
    List<DatanodeDescriptor> containingNodes;
    DatanodeDescriptor srcNode;
    INodeFile fileINode = null;

    int scheduledWork = 0;
    List<ReplicationWork> work = new LinkedList<ReplicationWork>();

    writeLock();
    try {
      synchronized (neededReplications) {
        for (priority = 0; priority < blocksToReplicate.size(); priority++) {
          for (BlockInfo block : blocksToReplicate.get(priority)) {
            // block should belong to a file
            BlockInfo storedBlock = blocksMap.getBlockInfo(block);
            fileINode = (storedBlock == null) ? null : storedBlock.getINode();
            // abandoned block not belong to a file
            if (fileINode == null) {
              neededReplications.remove(block, priority); // remove from neededReplications
              replIndex[priority]--;
              continue;
            }
            requiredReplication = fileINode.getBlockReplication(storedBlock);

            // get a source data-node
            containingNodes = new ArrayList<DatanodeDescriptor>();
            NumberReplicas numReplicas = new NumberReplicas();
            srcNode = chooseSourceDatanode(block, containingNodes, numReplicas);
            if (srcNode == null) // block can not be replicated from any node
            {
              continue;
            }

            // do not schedule more if enough replicas is already pending
            numEffectiveReplicas = numReplicas.liveReplicas() +
              pendingReplications.getNumReplicas(block);
            if (numEffectiveReplicas >= requiredReplication) {
              neededReplications.remove(block, priority); // remove from neededReplications
              replIndex[priority]--;
              continue;
            }
            work.add(new ReplicationWork(block, fileINode, requiredReplication
                - numEffectiveReplicas, srcNode, containingNodes, priority));
          }
        }
      }
    } finally {
      writeUnlock();
    }

    // choose replication targets: NOT HODING THE GLOBAL LOCK
    for(ReplicationWork rw : work){
      DatanodeDescriptor targets[] = chooseTarget(rw);
      rw.targets = targets;
    }

    writeLock();
    try {
      for(ReplicationWork rw : work){
        DatanodeDescriptor[] targets = rw.targets;
        if(targets == null || targets.length == 0){
          rw.targets = null;
          continue;
        }
        if (targets.length == 1 && 
            !clusterMap.isOnSameRack(rw.srcNode, targets[0])) {
          // in case src & target are not on the same rack
          // see if there is an alternate valid source in the same rack 
          // as the target that we can use instead.
          for (DatanodeDescriptor node : rw.containingNodes) {
            if (node != rw.srcNode && 
                clusterMap.isOnSameRack(node, targets[0]) &&
                isGoodReplica(node, rw.block)) {
              rw.srcNode = node;
              break;
            }
          }
        }
        synchronized (neededReplications) {
          BlockInfo block = rw.block;
          priority = rw.priority;
          // Recheck since global lock was released
          // block should belong to a file
          BlockInfo storedBlock = blocksMap.getBlockInfo(block);
          fileINode = (storedBlock == null) ? null : storedBlock.getINode();
          // abandoned block not belong to a file
          if (fileINode == null ) {
            neededReplications.remove(block, priority); // remove from neededReplications
            rw.targets = null;
            replIndex[priority]--;
            continue;
          }
          requiredReplication = fileINode.getBlockReplication(storedBlock);

          // do not schedule more if enough replicas is already pending
          NumberReplicas numReplicas = countNodes(block);
          numEffectiveReplicas = numReplicas.liveReplicas() +
            pendingReplications.getNumReplicas(block);
          if (numEffectiveReplicas >= requiredReplication) {
            neededReplications.remove(block, priority); // remove from neededReplications
            replIndex[priority]--;
            rw.targets = null;
            continue;
          }

          // Add block to the to be replicated list
          rw.srcNode.addBlockToBeReplicated(block, targets);
          
          scheduledWork++;

          for (DatanodeDescriptor dn : targets) {
            dn.incBlocksScheduled();
          }

          // Move the block-replication into a "pending" state.
          // The reason we use 'pending' is so we can retry
          // replications that fail after an appropriate amount of time.
          pendingReplications.add(block, targets.length);
          if (NameNode.stateChangeLog.isDebugEnabled()) {
            NameNode.stateChangeLog.debug(
              "BLOCK* block " + block
                + " is moved from neededReplications to pendingReplications");
          }

          // remove from neededReplications
          if (numEffectiveReplicas + targets.length >= requiredReplication) {
            neededReplications.remove(block, priority); // remove from neededReplications
            replIndex[priority]--;
          }
        }
      }
    } finally {
      writeUnlock();
    }

    // update metrics
    updateReplicationMetrics(work);
    
    // print debug information
    if(NameNode.stateChangeLog.isInfoEnabled()){
      // log which blocks have been scheduled for replication
      for(ReplicationWork rw : work){
        // report scheduled blocks
        DatanodeDescriptor[] targets = rw.targets;
        if (targets != null && targets.length != 0) {
          StringBuffer targetList = new StringBuffer("datanode(s)");
          for (int k = 0; k < targets.length; k++) {
            targetList.append(' ');
            targetList.append(targets[k].getName());
          }
          NameNode.stateChangeLog.info(
            "BLOCK* ask "
              + rw.srcNode.getName() + " to replicate "
              + rw.block + " to " + targetList);
        }
      }
    }

    // log once per call
    if (NameNode.stateChangeLog.isDebugEnabled()) {
      NameNode.stateChangeLog.debug("BLOCK* neededReplications = "
          + neededReplications.size() + " pendingReplications = "
          + pendingReplications.size());
    }
    return scheduledWork;
  }
393462
def unscheduleFirst(self, runnable):
        
        for evt in self.store.query(TimedEvent, TimedEvent.runnable == runnable, sort=TimedEvent.time.ascending):
            evt.deleteFromStore()
            break
1047318
def __process_non_finalizable_job(self, job_record, uow):
        
        if uow.is_active:
            # Large Job processing takes more than 1 tick of Scheduler
            # Let the Large Job processing complete - do no updates to Scheduler records
            pass
        elif uow.is_finished:
            # create new uow to cover new inserts
            uow, is_duplicate = self.insert_and_publish_uow(job_record, 0, int(uow.end_id) + 1)
            self.update_job(job_record, uow, job.STATE_IN_PROGRESS)
872584
private final ClientResponseImpl dispatchLoadSinglepartitionTable(Procedure catProc,
                                                        StoredProcedureInvocation task,
                                                        InvocationClientHandler handler,
                                                        Connection ccxn)
    {
        int partition = -1;
        try {
            CatalogMap<Table> tables = m_catalogContext.get().database.getTables();
            int partitionParamType = getLoadSinglePartitionTablePartitionParamType(tables, task);
            byte[] valueToHash = (byte[])task.getParameterAtIndex(0);
            partition = TheHashinator.getPartitionForParameter(partitionParamType, valueToHash);
        }
        catch (Exception e) {
            authLog.warn(e.getMessage());
            return new ClientResponseImpl(ClientResponseImpl.UNEXPECTED_FAILURE,
                                          new VoltTable[0], e.getMessage(), task.clientHandle);
        }
        assert(partition != -1);
        createTransaction(handler.connectionId(),
                          task,
                          catProc.getReadonly(),
                          catProc.getSinglepartition(),
                          catProc.getEverysite(),
                          new int[] { partition },
                          task.getSerializedSize(),
                          System.nanoTime());
        return null;
    }
1294382
def schedule_deleted(sender, instance, **kwargs):
    
    from contentstore.tasks import deactivate_schedule

    deactivate_schedule.delay(str(instance.scheduler_schedule_id))
106810
public void scheduleReconnect() {

        logger.debug("{} scheduleReconnect()", logPrefix());

        if (!isEventLoopGroupActive()) {
            logger.debug("isEventLoopGroupActive() == false");
            return;
        }

        if (!isListenOnChannelInactive()) {
            logger.debug("Skip reconnect scheduling, listener disabled");
            return;
        }

        if ((channel == null || !channel.isActive()) && reconnectSchedulerSync.compareAndSet(false, true)) {

            attempts++;
            final int attempt = attempts;
            int timeout = (int) reconnectDelay.createDelay(attempt).toMillis();
            logger.debug("{} Reconnect attempt {}, delay {}ms", logPrefix(), attempt, timeout);

            this.reconnectScheduleTimeout = timer.newTimeout(it -> {

                reconnectScheduleTimeout = null;

                if (!isEventLoopGroupActive()) {
                    logger.warn("Cannot execute scheduled reconnect timer, reconnect workers are terminated");
                    return;
                }

                reconnectWorkers.submit(() -> {
                    ConnectionWatchdog.this.run(attempt);
                    return null;
                });
            }, timeout, TimeUnit.MILLISECONDS);

            // Set back to null when ConnectionWatchdog#run runs earlier than reconnectScheduleTimeout's assignment.
            if (!reconnectSchedulerSync.get()) {
                reconnectScheduleTimeout = null;
            }
        } else {
            logger.debug("{} Skipping scheduleReconnect() because I have an active channel", logPrefix());
        }
    }
60083
synchronized void unschedule(String scheduledJobId) {
    try {
      quartzScheduler.deleteJob(new JobKey(scheduledJobId, SCHEDULED_JOB_GROUP));
    } catch (SchedulerException e) {
      String message = format("Error deleting ScheduledJob ''{0}''", scheduledJobId);
      LOG.error(message, e);
      throw new ScheduledJobException(message, e);
    }
  }
578192
def unschedule_next
      clock, (obj, relative_time) = next_with_clock
      if obj
        clock.unschedule(obj)
        [obj, relative_time]
      end
    end
852539
@Override
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay) {
        ScheduledExecutorService executor = getScheduledExecutor();
        try {
            return executor.scheduleWithFixedDelay(errorHandlingTask(task, true), 0, delay, TimeUnit.MILLISECONDS);
        } catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
        }
    }