Given a FeedParition and Plan create a Driver that will consume from the feed partition
and execute the plan.
@param feedPartition
@param plan
@return an uninitialized Driver
542119
public static Driver createDriver(FeedPartition feedPartition, Plan plan, MetricRegistry metricRegistry){
    populateFeedMetricInfo(plan, feedPartition, metricRegistry);
    OperatorDesc desc = plan.getRootOperator();
    Operator oper = buildOperator(desc, metricRegistry, plan.getName(), feedPartition);
    OffsetStorage offsetStorage = null;
    OffsetStorageDesc offsetDesc = plan.getOffsetStorageDesc();
    if (offsetDesc != null && feedPartition.supportsOffsetManagement()){
      offsetStorage = buildOffsetStorage(feedPartition, plan, offsetDesc);
      Offset offset = offsetStorage.findLatestPersistedOffset();
      if (offset != null){
        feedPartition.setOffset(new String(offset.serialize(), Charsets.UTF_8));
      }
    }
    CollectorProcessor cp = new CollectorProcessor();
    cp.setTupleRetry(plan.getTupleRetry());
    int offsetCommitInterval = plan.getOffsetCommitInterval();
    Driver driver = new Driver(feedPartition, oper, offsetStorage, cp, offsetCommitInterval, metricRegistry, plan.getName());
    recurseOperatorAndDriverNode(desc, driver.getDriverNode(), metricRegistry, feedPartition);
    return driver;
  }
784729
private <K> DataStream<T> partitionCustom(Partitioner<K> partitioner, Keys<T> keys) {
		KeySelector<T, K> keySelector = KeySelectorUtil.getSelectorForOneKey(keys, partitioner, getType(), getExecutionConfig());

		return setConnectionType(
				new CustomPartitionerWrapper<>(
						clean(partitioner),
						clean(keySelector)));
	}
592458
public static XContentBuilder marshall(PlanVersionBean bean) throws StorageException {
        try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
            PlanBean plan = bean.getPlan();
            OrganizationBean org = plan.getOrganization();
            preMarshall(bean);
            builder
                .startObject()
                    .field("organizationId", org.getId())
                    .field("organizationName", org.getName())
                    .field("planId", plan.getId())
                    .field("planName", plan.getName())
                    .field("planDescription", plan.getDescription())
                    .field("version", bean.getVersion())
                    .field("status", bean.getStatus())
                    .field("createdBy", bean.getCreatedBy())
                    .field("createdOn", bean.getCreatedOn().getTime())
                    .field("modifiedBy", bean.getModifiedBy())
                    .field("modifiedOn", bean.getModifiedOn().getTime())
                    .field("lockedOn", bean.getLockedOn() != null ? bean.getLockedOn().getTime() : null)
                .endObject();
            postMarshall(bean);
            return builder;
        } catch (IOException e) {
            throw new StorageException(e);
        }
    }
168069
def _partition(self):
        

        # KazooClient and SetPartitioner objects need to be instantiated after
        # the consumer process has forked.  Instantiating prior to forking
        # gives the appearance that things are working but after forking the
        # connection to zookeeper is lost and no state changes are visible

        if not self._kazoo_client:
            self._kazoo_client = KazooClient(hosts=self._zookeeper_url)
            self._kazoo_client.start()

            state_change_event = threading.Event()

            self._set_partitioner = (
                SetPartitioner(self._kazoo_client,
                               path=self._zookeeper_path,
                               set=self._consumer.fetch_offsets.keys(),
                               state_change_event=state_change_event,
                               identifier=str(datetime.datetime.now())))

        try:
            while 1:
                if self._set_partitioner.failed:
                    raise Exception("Failed to acquire partition")

                elif self._set_partitioner.release:
                    log.info("Releasing locks on partition set {} "
                             "for topic {}".format(self._partitions,
                                                   self._kafka_topic))
                    self._set_partitioner.release_set()

                    self._partitions = []

                elif self._set_partitioner.acquired:
                    if not self._partitions:
                        self._partitions = [p for p in self._set_partitioner]

                        if not self._partitions:
                            log.info("Not assigned any partitions on topic {},"
                                     " waiting for a Partitioner state change"
                                     .format(self._kafka_topic))
                            state_change_event.wait()
                            state_change_event.clear()
                            continue

                        log.info("Acquired locks on partition set {} "
                                 "for topic {}".format(self._partitions, self._kafka_topic))

                        # Reconstruct the kafka consumer object because the
                        # consumer has no API that allows the set of partitons
                        # to be updated outside of construction.
                        self._consumer.stop()
                        self._consumer = self._create_kafka_consumer(self._partitions)
                        return

                elif self._set_partitioner.allocating:
                    log.info("Waiting to acquire locks on partition set")
                    self._set_partitioner.wait_for_acquire()

        except Exception:
            log.exception('KafkaConsumer encountered fatal exception '
                          'processing messages.')
            raise
27552
public void execute() {
		Map<Partition, List<Node>> map = getNodesByPartition();
		for (Partition partition : map.keySet()) {
			List<Node> nodes = map.get(partition);
			partition.execute(nodes);
		}
	}
47409
private Configuration makeDriverConfiguration(
      final JobSubmissionEvent jobSubmissionEvent,
      final Path jobFolderPath) throws IOException {

    return this.driverConfigurationProvider.getDriverConfiguration(
            jobFolderPath.toUri(),
            jobSubmissionEvent.getRemoteId(),
            jobSubmissionEvent.getIdentifier(),
            jobSubmissionEvent.getConfiguration());
  }
1034775
@Override
	public void prepare() throws Exception {
		final TaskConfig config = this.taskContext.getTaskConfig();
		if (config.getDriverStrategy() != DriverStrategy.ALL_REDUCE) {
			throw new Exception("Unrecognized driver strategy for AllReduce driver: " + config.getDriverStrategy().name());
		}
		
		TypeSerializerFactory<T> serializerFactory = this.taskContext.getInputSerializer(0);
		this.serializer = serializerFactory.getSerializer();
		this.input = this.taskContext.getInput(0);

		ExecutionConfig executionConfig = taskContext.getExecutionConfig();
		this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();

		if (LOG.isDebugEnabled()) {
			LOG.debug("AllReduceDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
		}
	}
541544
@Override
	public void prepare() throws Exception {
		TaskConfig config = this.taskContext.getTaskConfig();
		if (config.getDriverStrategy() != DriverStrategy.SORTED_GROUP_REDUCE) {
			throw new Exception("Unrecognized driver strategy for GroupReduce driver: " + config.getDriverStrategy().name());
		}
		this.serializer = this.taskContext.<IT>getInputSerializer(0).getSerializer();
		this.comparator = this.taskContext.getInputComparator(0);
		this.input = this.taskContext.getInput(0);
	}
253741
public Map<Integer, Map<ResourceType, Integer>> getCpuToResourcePartitioning() {
    if (cachedCpuToResourcePartitioning == null) {
      cachedCpuToResourcePartitioning =
          getUncachedCpuToResourcePartitioning(this);
    }
    return cachedCpuToResourcePartitioning;
  }
1324678
public void addPartition(RedisClusterNode partition) {

        LettuceAssert.notNull(partition, "Partition must not be null");

        synchronized (partitions) {
            slotCache = EMPTY;
            partitions.add(partition);
        }
    }
291330
@VisibleForTesting
	@Override
	public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
		if (partitionRequestClient == null) {
			// Create a client and request the partition
			partitionRequestClient = connectionManager
				.createPartitionRequestClient(connectionId);

			partitionRequestClient.requestSubpartition(partitionId, subpartitionIndex, this, 0);
		}
	}
257754
public static ResultPartitionDeploymentDescriptor from(
			IntermediateResultPartition partition, int maxParallelism, boolean lazyScheduling) {

		final IntermediateDataSetID resultId = partition.getIntermediateResult().getId();
		final IntermediateResultPartitionID partitionId = partition.getPartitionId();
		final ResultPartitionType partitionType = partition.getIntermediateResult().getResultType();

		// The produced data is partitioned among a number of subpartitions.
		//
		// If no consumers are known at this point, we use a single subpartition, otherwise we have
		// one for each consuming sub task.
		int numberOfSubpartitions = 1;

		if (!partition.getConsumers().isEmpty() && !partition.getConsumers().get(0).isEmpty()) {

			if (partition.getConsumers().size() > 1) {
				throw new IllegalStateException("Currently, only a single consumer group per partition is supported.");
			}

			numberOfSubpartitions = partition.getConsumers().get(0).size();
		}

		return new ResultPartitionDeploymentDescriptor(
				resultId, partitionId, partitionType, numberOfSubpartitions, maxParallelism, lazyScheduling);
	}
876972
public static Feed buildFeed(FeedDesc feedDesc){
    Feed feed = null;
    NitFactory nitFactory = new NitFactory();
    NitDesc nitDesc = nitDescFromDynamic(feedDesc);
    nitDesc.setConstructorParameters(new Class [] { Map.class });
    nitDesc.setConstructorArguments(new Object[] { feedDesc.getProperties() });
    try {
      feed = nitFactory.construct(nitDesc);
    } catch (NitException e) {
      throw new RuntimeException(e);
    }
    return feed;
  }
581498
public synchronized CompiledPlan planSqlCore(String sql, StatementPartitioning partitioning) {
        TrivialCostModel costModel = new TrivialCostModel();
        DatabaseEstimates estimates = new DatabaseEstimates();

        CompiledPlan plan = null;
        // This try-with-resources block acquires a global lock on all planning
        // This is required until we figure out how to do parallel planning.
        try (QueryPlanner planner = new QueryPlanner(
                sql, "PlannerTool", "PlannerToolProc", m_database,
                partitioning, m_hsql, estimates, !VoltCompiler.DEBUG_MODE,
                costModel, null, null, DeterminismMode.FASTER, false)) {

            // do the expensive full planning.
            planner.parse();
            plan = planner.plan();
            assert(plan != null);
        }
        catch (Exception e) {
            /*
             * Don't log PlanningErrorExceptions or HSQLParseExceptions, as they
             * are at least somewhat expected.
             */
            String loggedMsg = "";
            if (!(e instanceof PlanningErrorException || e instanceof HSQLParseException)) {
                logException(e, "Error compiling query");
                loggedMsg = " (Stack trace has been written to the log.)";
            }
            if (e.getMessage() != null) {
                throw new RuntimeException("SQL error while compiling query: " + e.getMessage() + loggedMsg, e);
            }
            throw new RuntimeException("SQL error while compiling query: " + e.toString() + loggedMsg, e);
        }

        if (plan == null) {
            throw new RuntimeException("Null plan received in PlannerTool.planSql");
        }

        return plan;
    }
1060662
protected DataStream<T> setConnectionType(StreamPartitioner<T> partitioner) {
		return new DataStream<>(this.getExecutionEnvironment(), new PartitionTransformation<>(this.getTransformation(), partitioner));
	}
1240612
void onConsumedPartition(ResultPartition partition) {
		final ResultPartition previous;

		LOG.debug("Received consume notification from {}.", partition);

		synchronized (registeredPartitions) {
			previous = registeredPartitions.remove(partition.getPartitionId());
		}

		// Release the partition if it was successfully removed
		if (partition == previous) {
			partition.release();

			LOG.debug("Released {}.", partition);
		}
	}
1205410
@Override
    public synchronized ServoDriver getServoDriver(Pin servoPin) throws IOException {
        List<Pin> servoPins = getDefinedServoPins();
        int index = servoPins.indexOf(servoPin);
        if (index < 0) {
            throw new IllegalArgumentException("Servo driver cannot drive pin " + servoPin);
        }

        RPIServoBlasterServoDriver driver = allocatedDrivers.get(servoPin);
        if (driver == null) {
            driver = new RPIServoBlasterServoDriver(servoPin, index, PIN_MAP.get(servoPin), this);
            ensureWriterIsCreated();
        }

        return driver;
    }