Alternative to parallelStreams() which executes a runnable in a separate
pool.

@param runnable
@param concurrencyConfiguration
@param parallelStream
339424
public static void forkJoinExecution(Runnable runnable, ConcurrencyConfiguration concurrencyConfiguration, boolean parallelStream) {
        if(parallelStream && concurrencyConfiguration.isParallelized()) {
            try {
                ForkJoinPool pool = new ForkJoinPool(concurrencyConfiguration.getMaxNumberOfThreadsPerTask());
                pool.submit(runnable).get();
                pool.shutdown();
            } 
            catch (InterruptedException | ExecutionException ex) {
                throw new RuntimeException(ex);
            }
        }
        else {
            runnable.run();
        }
    }
1117887
public void perform(TaskRequest req, TaskResponse res) {
         final String poolAttributeName = (String) this.attributeNamePhrase.evaluate(req, res);
         final boolean useExisting = Boolean.valueOf((String) this.useExistingPhrase.evaluate(req, res));
         
         final ExecutorService executorService;
         final boolean usingExisting;
         if (useExisting && req.hasAttribute(poolAttributeName)) {
             // Try using an existing pool
             executorService = (ExecutorService) req.getAttribute(poolAttributeName);
             usingExisting = true;
         }
         else {
             //No existing pool or configured to ignore it, create a new pool
             final int threads = Integer.parseInt((String) this.threadsPhrase.evaluate(req, res));
             final int queueRatio = Integer.parseInt((String) this.queueRatioPhrase.evaluate(req, res));
             
             if (this.log.isDebugEnabled()) {
                 this.log.debug("Creating ThreadPoolExecutor with " + threads + " threads and queue space of " + (threads * queueRatio));
             }
             
             executorService = this.createExecutorService(threads, queueRatio);

             usingExisting = false;
         }
         
         try {
             //Provide the pool as a response attribute
             res.setAttribute(poolAttributeName, executorService);
             
             //Delegate to sub-task execution
             this.performSubtasks(req, res);
         }
         finally {
             //If using an existing pool it should not get shutdown here
             if (usingExisting) {
                 return;
             }
             
             //If shutdown already it is due to a child task, likely from a fail-fast ConcurrentTask handling an exception
             if (executorService.isShutdown()) {
                 throw new IllegalStateException("The ExecutorService was shutdown by a child task. This is likely due to an exception inside of a ConcurrentTask. Check the logs for information about the failed ConcurrentTask");
             }
             
             //Cleanly shutdown the pool
             executorService.shutdown();
             
             //Wait for all tasks to complete
             try {
                 executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
             }
             catch (InterruptedException e) {
                 this.log.warn("Interrupted while waiting for thread pool to shutdown", e);
             }
         }
     }
1062275
public double sum(DoubleStream stream) {
        Callable<Double> callable = () -> stream.sum();
        return ThreadMethods.forkJoinExecution(callable, concurrencyConfiguration, stream.isParallel());
    }
60363
public CassandraSink<IN> setParallelism(int parallelism) {
		if (useDataStreamSink) {
			getSinkTransformation().setParallelism(parallelism);
		} else {
			getStreamTransformation().setParallelism(parallelism);
		}
		return this;
	}
376797
public void startNewThread(Runnable runnable) throws InterruptedException {
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.entry(tc, "startNewThread");

        if (_consumerThreadPool == null) {
            createConsumerThreadPool();
        }
        try {
            _consumerThreadPool.execute(runnable,
                                        ThreadPool.EXPAND_WHEN_QUEUE_IS_FULL_WAIT_AT_LIMIT);
        } catch (ThreadPoolQueueIsFullException e) {
            // Exception should never occur as we are waiting when the queue is
            // full.
            FFDCFilter
                            .processException(
                                              e,
                                              "com.ibm.ws.sib.processor.impl.MessageProcessor.startNewThread",
                                              "1:2705:1.445", this);
        }

        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.exit(tc, "startNewThread");
    }
1059785
public ConfigurationBuilder useParallelExecutor(final int availableProcessors) {
        ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("org.reflections-scanner-%d").build();
        setExecutorService(Executors.newFixedThreadPool(availableProcessors, factory));
        return this;
    }
1298134
public <T> void forEach(Stream<T> stream, Consumer<? super T> action) {
        Runnable runnable = () -> stream.forEach(action);
        ThreadMethods.forkJoinExecution(runnable, concurrencyConfiguration, stream.isParallel());
    }
28446
public void runnable(Runnable runnable) {
		initStartTotal();
		long startInvokation = System.currentTimeMillis();
		runnable.run();
		invoked(System.currentTimeMillis() - startInvokation);
	}
1362729
public static <T> Stream<T> iteratorToStream(final Iterator<T> iterator, final Boolean parallel) {
        return stream(spliteratorUnknownSize(iterator, IMMUTABLE), parallel);
    }
1304952
public void runFetcher() throws Exception {

		// check that we are running before proceeding
		if (!running) {
			return;
		}

		this.mainThread = Thread.currentThread();

		// ------------------------------------------------------------------------
		//  Procedures before starting the infinite while loop:
		// ------------------------------------------------------------------------

		//  1. check that there is at least one shard in the subscribed streams to consume from (can be done by
		//     checking if at least one value in subscribedStreamsToLastDiscoveredShardIds is not null)
		boolean hasShards = false;
		StringBuilder streamsWithNoShardsFound = new StringBuilder();
		for (Map.Entry<String, String> streamToLastDiscoveredShardEntry : subscribedStreamsToLastDiscoveredShardIds.entrySet()) {
			if (streamToLastDiscoveredShardEntry.getValue() != null) {
				hasShards = true;
			} else {
				streamsWithNoShardsFound.append(streamToLastDiscoveredShardEntry.getKey()).append(", ");
			}
		}

		if (streamsWithNoShardsFound.length() != 0 && LOG.isWarnEnabled()) {
			LOG.warn("Subtask {} has failed to find any shards for the following subscribed streams: {}",
				indexOfThisConsumerSubtask, streamsWithNoShardsFound.toString());
		}

		if (!hasShards) {
			throw new RuntimeException("No shards can be found for all subscribed streams: " + streams);
		}

		//  2. start consuming any shard state we already have in the subscribedShardState up to this point; the
		//     subscribedShardState may already be seeded with values due to step 1., or explicitly added by the
		//     consumer using a restored state checkpoint
		for (int seededStateIndex = 0; seededStateIndex < subscribedShardsState.size(); seededStateIndex++) {
			KinesisStreamShardState seededShardState = subscribedShardsState.get(seededStateIndex);

			// only start a consuming thread if the seeded subscribed shard has not been completely read already
			if (!seededShardState.getLastProcessedSequenceNum().equals(SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get())) {

				if (LOG.isInfoEnabled()) {
					LOG.info("Subtask {} will start consuming seeded shard {} from sequence number {} with ShardConsumer {}",
						indexOfThisConsumerSubtask, seededShardState.getStreamShardHandle().toString(),
						seededShardState.getLastProcessedSequenceNum(), seededStateIndex);
					}

				shardConsumersExecutor.submit(
					createShardConsumer(
						seededStateIndex,
						subscribedShardsState.get(seededStateIndex).getStreamShardHandle(),
						subscribedShardsState.get(seededStateIndex).getLastProcessedSequenceNum(),
						registerShardMetrics(consumerMetricGroup, subscribedShardsState.get(seededStateIndex))));
			}
		}

        // start periodic watermark emitter, if a watermark assigner was configured
		if (periodicWatermarkAssigner != null) {
			long periodicWatermarkIntervalMillis = runtimeContext.getExecutionConfig().getAutoWatermarkInterval();
			if (periodicWatermarkIntervalMillis > 0) {
				ProcessingTimeService timerService = ((StreamingRuntimeContext) runtimeContext).getProcessingTimeService();
				LOG.info("Starting periodic watermark emitter with interval {}", periodicWatermarkIntervalMillis);
				new PeriodicWatermarkEmitter(timerService, periodicWatermarkIntervalMillis).start();
			}
			this.shardIdleIntervalMillis = Long.parseLong(
				getConsumerConfiguration().getProperty(ConsumerConfigConstants.SHARD_IDLE_INTERVAL_MILLIS,
					Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_IDLE_INTERVAL_MILLIS)));
		}

		// ------------------------------------------------------------------------

		// finally, start the infinite shard discovery and consumer launching loop;
		// we will escape from this loop only when shutdownFetcher() or stopWithError() is called

		final long discoveryIntervalMillis = Long.valueOf(
			configProps.getProperty(
				ConsumerConfigConstants.SHARD_DISCOVERY_INTERVAL_MILLIS,
				Long.toString(ConsumerConfigConstants.DEFAULT_SHARD_DISCOVERY_INTERVAL_MILLIS)));

		if (this.numberOfActiveShards.get() == 0) {
			LOG.info("Subtask {} has no active shards to read on startup; marking the subtask as temporarily idle ...",
				indexOfThisConsumerSubtask);
			sourceContext.markAsTemporarilyIdle();
		}

		while (running) {
			if (LOG.isDebugEnabled()) {
				LOG.debug("Subtask {} is trying to discover new shards that were created due to resharding ...",
					indexOfThisConsumerSubtask);
			}
			List<StreamShardHandle> newShardsDueToResharding = discoverNewShardsToSubscribe();

			for (StreamShardHandle shard : newShardsDueToResharding) {
				// since there may be delay in discovering a new shard, all new shards due to
				// resharding should be read starting from the earliest record possible
				KinesisStreamShardState newShardState =
					new KinesisStreamShardState(convertToStreamShardMetadata(shard), shard, SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get());
				int newStateIndex = registerNewSubscribedShardState(newShardState);

				if (LOG.isInfoEnabled()) {
					LOG.info("Subtask {} has discovered a new shard {} due to resharding, and will start consuming " +
							"the shard from sequence number {} with ShardConsumer {}",
						indexOfThisConsumerSubtask, newShardState.getStreamShardHandle().toString(),
						newShardState.getLastProcessedSequenceNum(), newStateIndex);
				}

				shardConsumersExecutor.submit(
					createShardConsumer(
						newStateIndex,
						newShardState.getStreamShardHandle(),
						newShardState.getLastProcessedSequenceNum(),
						registerShardMetrics(consumerMetricGroup, newShardState)));
			}

			// we also check if we are running here so that we won't start the discovery sleep
			// interval if the running flag was set to false during the middle of the while loop
			if (running && discoveryIntervalMillis != 0) {
				try {
					Thread.sleep(discoveryIntervalMillis);
				} catch (InterruptedException iex) {
					// the sleep may be interrupted by shutdownFetcher()
				}
			}
		}

		// make sure all resources have been terminated before leaving
		try {
			awaitTermination();
		} catch (InterruptedException ie) {
			// If there is an original exception, preserve it, since that's more important/useful.
			this.error.compareAndSet(null, ie);
		}

		// any error thrown in the shard consumer threads will be thrown to the main thread
		Throwable throwable = this.error.get();
		if (throwable != null) {
			if (throwable instanceof Exception) {
				throw (Exception) throwable;
			} else if (throwable instanceof Error) {
				throw (Error) throwable;
			} else {
				throw new Exception(throwable);
			}
		}
	}
694424
public static <T> Stream<T> stream(Stream<T> stream, boolean parallel) {
        if(parallel) {
            return stream.parallel();
        }
        else {
            return stream.sequential();
        }
    }
960104
public void run(Runnable runnable, Executor executor)
    {
        if (this.goForward())
        {
            for (int i = this.from; i <= this.to; i += this.step)
            {
                executor.execute(runnable);
            }
        }
        else
        {
            for (int i = this.from; i >= this.to; i += this.step)
            {
                executor.execute(runnable);
            }
        }
    }
1107271
@Override
  public void execute(Runnable runnable) {
    if (runnable == null) {
      throw new NullPointerException("runnable parameter is null");
    }

    if (!mWorkQueue.offer(runnable)) {
      throw new RejectedExecutionException(
          mName + " queue is full, size=" + mWorkQueue.size());
    }

    final int queueSize = mWorkQueue.size();
    final int maxSize = mMaxQueueSize.get();
    if ((queueSize > maxSize) && mMaxQueueSize.compareAndSet(maxSize, queueSize)) {
      FLog.v(TAG, "%s: max pending work in queue = %d", mName, queueSize);
    } // else, there was a race and another thread updated and logged the max queue size

    startWorkerIfNeeded();
  }
121618
public void doRuntimeConfiguration(RuntimeSpringConfiguration springConfig) {
        ApplicationContext context = springConfig.getUnrefreshedApplicationContext();
        AutowireCapableBeanFactory autowireCapableBeanFactory = context.getAutowireCapableBeanFactory();
        if(autowireCapableBeanFactory instanceof ConfigurableListableBeanFactory) {
            ConfigurableListableBeanFactory beanFactory = (ConfigurableListableBeanFactory)autowireCapableBeanFactory;
            ConversionService existingConversionService = beanFactory.getConversionService();
            ConverterRegistry converterRegistry;
            if(existingConversionService == null) {
                GenericConversionService conversionService = new GenericConversionService();
                converterRegistry = conversionService;
                beanFactory.setConversionService(conversionService);
            }
            else {
                converterRegistry = (ConverterRegistry)existingConversionService;
            }

            converterRegistry.addConverter(new Converter<NavigableMap.NullSafeNavigator, Object>() {
                @Override
                public Object convert(NavigableMap.NullSafeNavigator source) {
                    return null;
                }
            });
        }
        checkInitialised();
        for (GrailsPlugin plugin : pluginList) {
            if (plugin.supportsCurrentScopeAndEnvironment() && plugin.isEnabled(context.getEnvironment().getActiveProfiles())) {
                plugin.doWithRuntimeConfiguration(springConfig);
            }
        }
    }
751928
public static <T> Stream<T> buildStream(boolean isParallel,
            Collection<T> collection) {
        return isParallel ? collection.parallelStream() : collection.stream();
    }
284549
public static void runIntoJTPSync(final JRebirthRunnable runnable, final long... timeout) {
        final SyncRunnable sync = new SyncRunnable(runnable);

        if (JRebirth.isJTPSlot()) {
            // We are into a JTP slot so just run it synchronously
            sync.run();
            // Be careful in this case no timeout protection is achieved
        } else {
            // The runnable will be run into the JTP slot during the next round
            JRebirthThread.getThread().runIntoJTP(sync);
            // Wait the end of the runnable execution
            sync.waitEnd(timeout);
        }
    }
76937
public static void invokeOnEventDispatchThread(Runnable runnable)
    {
        if (SwingUtilities.isEventDispatchThread())
        {
            runnable.run();
        }
        else
        {
            SwingUtilities.invokeLater(runnable);
        }
    }