{@inheritDoc}

<p>
If this stream was created using {@link #parallel(ForkJoinPool)}, the new
stream forgets about supplied custom {@link ForkJoinPool} and its
terminal operation will be executed in common pool.
976652
@SuppressWarnings("unchecked")
    @Override
    public S parallel() {
        context = context.parallel();
        if (stream != null)
            stream = stream.parallel();
        return (S) this;
    }
1116887
public static ExecutorService newSingleThreadPoolExecutor() {
        ExecutorService singleThreadPoolExecutor = Executors.newSingleThreadExecutor();
        EXECUTORS.add(singleThreadPoolExecutor);
        return singleThreadPoolExecutor;
    }
450458
public void execute(Fork fork) {
    LOG.info(String.format("Executing fork %d of task %s", fork.getIndex(), fork.getTaskId()));
    this.forkExecutor.execute(fork);
  }
802009
public ParallelTask execute(ParallecResponseHandler handler) {

        ParallelTask task = new ParallelTask();

        try {
            targetHostMeta = new TargetHostMeta(targetHosts);

            final ParallelTask taskReal = new ParallelTask(requestProtocol,
                    concurrency, httpMeta, targetHostMeta, sshMeta, tcpMeta, udpMeta, pingMeta,
                    handler, responseContext, 
                    replacementVarMapNodeSpecific, replacementVarMap,
                    requestReplacementType, 
                    config);

            task = taskReal;

            logger.info("***********START_PARALLEL_HTTP_TASK_"
                    + task.getTaskId() + "***********");

            // throws ParallelTaskInvalidException
            task.validateWithFillDefault();

            task.setSubmitTime(System.currentTimeMillis());

            if (task.getConfig().isEnableCapacityAwareTaskScheduler()) {

                //late initialize the task scheduler
                ParallelTaskManager.getInstance().initTaskSchedulerIfNot();
                // add task to the wait queue
                ParallelTaskManager.getInstance().getWaitQ().add(task);

                logger.info("Enabled CapacityAwareTaskScheduler. Submitted task to waitQ in builder.. "
                        + task.getTaskId());

            } else {

                logger.info(
                        "Disabled CapacityAwareTaskScheduler. Immediately execute task {} ",
                        task.getTaskId());

                Runnable director = new Runnable() {

                    public void run() {
                        ParallelTaskManager.getInstance()
                                .generateUpdateExecuteTask(taskReal);
                    }
                };
                new Thread(director).start();
            }

            if (this.getMode() == TaskRunMode.SYNC) {
                logger.info("Executing task {} in SYNC mode...  ",
                        task.getTaskId());

                while (task != null && !task.isCompleted()) {
                    try {
                        Thread.sleep(500L);
                    } catch (InterruptedException e) {
                        logger.error("InterruptedException " + e);
                    }
                }
            }
        } catch (ParallelTaskInvalidException ex) {

            logger.info("Request is invalid with missing parts. Details: "
                    + ex.getMessage() + " Cannot execute at this time. "
                    + " Please review your request and try again.\nCommand:"
                    + httpMeta.toString());

            task.setState(ParallelTaskState.COMPLETED_WITH_ERROR);
            task.getTaskErrorMetas().add(
                    new TaskErrorMeta(TaskErrorType.VALIDATION_ERROR,
                            "validation eror"));

        } catch (Exception t) {
            logger.error("fail task builder. Unknown error: " + t, t);
            task.setState(ParallelTaskState.COMPLETED_WITH_ERROR);
            task.getTaskErrorMetas().add(
                    new TaskErrorMeta(TaskErrorType.UNKNOWN, "unknown eror",
                            t));
        }

        logger.info("***********FINISH_PARALLEL_HTTP_TASK_" + task.getTaskId()
                + "***********");
        return task;

    }
1126764
@Override
	protected void run() throws Exception {

		final String iterationId = getConfiguration().getIterationId();
		if (iterationId == null || iterationId.length() == 0) {
			throw new Exception("Missing iteration ID in the task configuration");
		}

		final String brokerID = createBrokerIdString(getEnvironment().getJobID(), iterationId ,
				getEnvironment().getTaskInfo().getIndexOfThisSubtask());

		final long iterationWaitTime = getConfiguration().getIterationWaitTime();
		final boolean shouldWait = iterationWaitTime > 0;

		final BlockingQueue<StreamRecord<OUT>> dataChannel = new ArrayBlockingQueue<StreamRecord<OUT>>(1);

		// offer the queue for the tail
		BlockingQueueBroker.INSTANCE.handIn(brokerID, dataChannel);
		LOG.info("Iteration head {} added feedback queue under {}", getName(), brokerID);

		// do the work
		try {
			@SuppressWarnings("unchecked")
			RecordWriterOutput<OUT>[] outputs = (RecordWriterOutput<OUT>[]) getStreamOutputs();

			// If timestamps are enabled we make sure to remove cyclic watermark dependencies
			if (isSerializingTimestamps()) {
				synchronized (getCheckpointLock()) {
					for (RecordWriterOutput<OUT> output : outputs) {
						output.emitWatermark(new Watermark(Long.MAX_VALUE));
					}
				}
			}

			while (running) {
				StreamRecord<OUT> nextRecord = shouldWait ?
					dataChannel.poll(iterationWaitTime, TimeUnit.MILLISECONDS) :
					dataChannel.take();

				if (nextRecord != null) {
					synchronized (getCheckpointLock()) {
						for (RecordWriterOutput<OUT> output : outputs) {
							output.collect(nextRecord);
						}
					}
				}
				else {
					// done
					break;
				}
			}
		}
		finally {
			// make sure that we remove the queue from the broker, to prevent a resource leak
			BlockingQueueBroker.INSTANCE.remove(brokerID);
			LOG.info("Iteration head {} removed feedback queue under {}", getName(), brokerID);
		}
	}
616678
public static <T> Stream<T> parallelConcatt(final Collection<? extends Iterator<? extends T>> c) {
        return parallelConcatt(c, DEFAULT_READING_THREAD_NUM);
    }
412832
public static <T, R> Stream<R> parallelZipp(final Collection<? extends Iterator<? extends T>> c, final Function<? super List<? extends T>, R> zipFunction) {
        return parallelZipp(c, zipFunction, DEFAULT_QUEUE_SIZE_PER_ITERATOR);
    }
89821
static WorkQueue commonSubmitterQueue() {
        ForkJoinPool p = common;
        int r = ThreadLocalRandom.getProbe();
        WorkQueue[] ws; int wl;
        return (p != null && (ws = p.workQueues) != null &&
                (wl = ws.length) > 0) ?
            ws[(wl - 1) & r & SQMASK] : null;
    }
1183878
@Nullable
    public static ForkJoinWorkerThreadFactory unwrap(@Nullable ForkJoinWorkerThreadFactory threadFactory) {
        if (!isDisableInheritableForkJoinWorkerThreadFactory(threadFactory)) return threadFactory;

        return ((DisableInheritableForkJoinWorkerThreadFactoryWrapper) threadFactory).unwrap();
    }
778611
public static ThreadPoolExecutor createDefaultExecutorService() {
        ThreadFactory threadFactory = new ThreadFactory() {
            private int threadCount = 1;

            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("s3-transfer-manager-worker-" + threadCount++);
                return thread;
            }
        };
        return (ThreadPoolExecutor)Executors.newFixedThreadPool(10, threadFactory);
    }
1358637
synchronized void execute(File root, Runnable task) {
    if (executors == null) {
      throw new RuntimeException("AsyncDiskService is already shutdown");
    }
    ThreadPoolExecutor executor = executors.get(root);
    if (executor == null) {
      throw new RuntimeException("Cannot find root " + root
          + " for execution of task " + task);
    } else {
      executor.execute(task);
    }
  }
890775
static Executor screenExecutor(Executor e) {
    if (!USE_COMMON_POOL && e == ForkJoinPool.commonPool())
      return ASYNC_POOL;
    return Objects.requireNonNull(e);
  }
383170
public static ThreadPoolExecutor newCachedThreadPool(int corePoolSize,
                                                         int maximumPoolSize,
                                                         BlockingQueue<Runnable> queue) {
        return new ThreadPoolExecutor(corePoolSize,
            maximumPoolSize,
            DateUtils.MILLISECONDS_PER_MINUTE,
            TimeUnit.MILLISECONDS,
            queue);
    }
1284606
@Override
	public JobExecutionResult execute(String jobName) throws Exception {
		// transform the streaming program into a JobGraph
		StreamGraph streamGraph = getStreamGraph();
		streamGraph.setJobName(jobName);

		JobGraph jobGraph = streamGraph.getJobGraph();
		jobGraph.setAllowQueuedScheduling(true);

		Configuration configuration = new Configuration();
		configuration.addAll(jobGraph.getJobConfiguration());
		configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0");

		// add (and override) the settings with what the user defined
		configuration.addAll(this.configuration);

		if (!configuration.contains(RestOptions.BIND_PORT)) {
			configuration.setString(RestOptions.BIND_PORT, "0");
		}

		int numSlotsPerTaskManager = configuration.getInteger(TaskManagerOptions.NUM_TASK_SLOTS, jobGraph.getMaximumParallelism());

		MiniClusterConfiguration cfg = new MiniClusterConfiguration.Builder()
			.setConfiguration(configuration)
			.setNumSlotsPerTaskManager(numSlotsPerTaskManager)
			.build();

		if (LOG.isInfoEnabled()) {
			LOG.info("Running job on local embedded Flink mini cluster");
		}

		MiniCluster miniCluster = new MiniCluster(cfg);

		try {
			miniCluster.start();
			configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().get().getPort());

			return miniCluster.executeJobBlocking(jobGraph);
		}
		finally {
			transformations.clear();
			miniCluster.close();
		}
	}
1157220
private int doJoin() {
        int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
        if ((s = status) >= 0) {
            if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) {
                if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue).
                    tryUnpush(this) || (s = doExec()) >= 0)
                    s = wt.pool.awaitJoin(w, this);
            }
            else
                s = externalAwaitDone();
        }
        return s;
    }
957532
public Stream<T> tailStream(T key, boolean inclusive, boolean parallel)
    {
        return StreamSupport.stream(tailSpliterator(key, inclusive), parallel);
    }
646288
public void purgeThreadLocal() {
        // if thread local pools exist, check to see if we have one created and
        // therefore need to tell it to purge
        if (null != threadLocals) {
            LocalThreadObjectPool pool = threadLocals.get();
            if (null != pool) {
                Object[] data = pool.purge();
                if (0 < data.length) {
                    mainPool.putBatch(data);
                }
            }
        }
    }