This demo runs a pipeline across ten different keys. The pipeline leverages
the hash value {bar} to determine the node where to send the data.

@throws Exception
507649
public void runPipelineWithHashtag() throws Exception {

		DynoJedisPipeline pipeline = client.pipelined();
		try {

			pipeline.set("pipeline-hashtag1-{bar}", "value-1");
			pipeline.set("pipeline-hashtag2-{bar}", "value-2");
			pipeline.set("pipeline-hashtag3-{bar}", "value-3");
			pipeline.set("pipeline-hashtag4-{bar}", "value-4");
			pipeline.set("pipeline-hashtag5-{bar}", "value-5");
			pipeline.set("pipeline-hashtag6-{bar}", "value-6");
			pipeline.set("pipeline-hashtag7-{bar}", "value-7");
			pipeline.set("pipeline-hashtag8-{bar}", "value-8");
			pipeline.set("pipeline-hashtag9-{bar}", "value-9");
			pipeline.set("pipeline-hashtag10-{bar}", "value-10");

			Response<String> value1 = pipeline.get("pipeline-hashtag1-{bar}");
			Response<String> value2 = pipeline.get("pipeline-hashtag2-{bar}");
			Response<String> value3 = pipeline.get("pipeline-hashtag3-{bar}");
			Response<String> value4 = pipeline.get("pipeline-hashtag4-{bar}");
			Response<String> value5 = pipeline.get("pipeline-hashtag5-{bar}");
			Response<String> value6 = pipeline.get("pipeline-hashtag6-{bar}");
			pipeline.sync();

			System.out.println(value1.get());
			System.out.println(value2.get());
			System.out.println(value3.get());
			System.out.println(value4.get());
			System.out.println(value5.get());
			System.out.println(value6.get());
		} catch (Exception e) {
			pipeline.discardPipelineAndReleaseConnection();
			throw e;
		}
	}
992880
public function run()
    {
        if (!$this->isAppRootSet()) {
            throw new Exception("The application root wasn't defined.", 1);
        }
        if (!$this->isConfigFileSet()) {
            throw new Exception("The main config file wasn't defined.", 2);
        }
        $sConfigPath = $this->getAppRoot().$this->getConfigFile();
        if (!is_readable($sConfigPath)) {
            throw new Exception("It's unable to load ".$sConfigPath
            .'as main config file.', 4);
        }
        $mConfig = include $sConfigPath;
        if (!is_array($mConfig)) {
            throw new Exception('The main config must be an array.', 5);
        }
        $this->configSet = $mConfig;
    }
402735
public void run() throws Exception {

        CmsObject cms = getCmsObject();
        String seoFilePath = cms.getRequestContext().getUri();
        CmsResource seoFile = cms.readResource(seoFilePath);
        m_configuration = new CmsXmlSeoConfiguration();
        m_configuration.load(cms, seoFile);
        String mode = m_configuration.getMode();
        if (mode.equals(CmsXmlSeoConfiguration.MODE_ROBOTS_TXT)) {
            showRobotsTxt();
        } else {
            boolean updateCache = Boolean.parseBoolean(getRequest().getParameter("updateCache"));
            String value = "";
            if (updateCache && m_configuration.usesCache()) {
                // update request, and caching is configured -> update the cache
                CmsXmlSitemapGenerator generator = prepareSitemapGenerator(seoFile, m_configuration);
                value = generator.renderSitemap();
                CmsXmlSitemapCache.INSTANCE.put(seoFile.getRootPath(), value);
            } else if (!updateCache && m_configuration.usesCache()) {
                // normal request, and caching is configured -> look in the cache first, and if not found, calculate sitemap and store it in cache
                String cachedValue = CmsXmlSitemapCache.INSTANCE.get(seoFile.getRootPath());
                if (cachedValue != null) {
                    value = cachedValue;
                } else {
                    CmsXmlSitemapGenerator generator = prepareSitemapGenerator(seoFile, m_configuration);
                    value = generator.renderSitemap();
                    CmsXmlSitemapCache.INSTANCE.put(seoFile.getRootPath(), value);
                }
            } else if (!updateCache && !m_configuration.usesCache()) {
                // normal request, caching is not configured -> always generate a fresh sitemap
                CmsXmlSitemapGenerator generator = prepareSitemapGenerator(seoFile, m_configuration);
                value = generator.renderSitemap();
            } else if (updateCache && !m_configuration.usesCache()) {
                // update request with no caching configured -> ignore
            }
            getResponse().getWriter().print(value);
        }

    }
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);
			}
		}
	}
39775
@Override
    public Boolean call() throws Exception {
        error.set(null);
        
        List<Callable<Boolean>> subtasks = Lists.newArrayList();
        
        // We are iterating the entire ring using an arbitrary number of threads
        if (this.concurrencyLevel != null || startToken != null|| endToken != null) {
            List<TokenRange> tokens = partitioner.splitTokenRange(
                    startToken == null ? partitioner.getMinToken() : startToken, 
                    endToken == null   ? partitioner.getMinToken() : endToken, 
                    this.concurrencyLevel == null ? 1 : this.concurrencyLevel);
            
            for (TokenRange range : tokens) {
                subtasks.add(makeTokenRangeTask(range.getStartToken(), range.getEndToken()));
            }
        }
        // We are iterating through each token range
        else {
            List<TokenRange> ranges = keyspace.describeRing(dc, rack);
            for (TokenRange range : ranges) {
                if (range.getStartToken().equals(range.getEndToken())) 
                    subtasks.add(makeTokenRangeTask(range.getStartToken(), range.getEndToken()));
                else
                    subtasks.add(makeTokenRangeTask(partitioner.getTokenMinusOne(range.getStartToken()), range.getEndToken()));
            }
        }
        
        try {
            // Use a local executor
            if (executor == null) {
                ExecutorService localExecutor = Executors
                        .newFixedThreadPool(subtasks.size(),
                            new ThreadFactoryBuilder().setDaemon(true)
                                .setNameFormat("AstyanaxAllRowsReader-%d")
                                .build());
                
                try {
                    futures.addAll(startTasks(localExecutor, subtasks));
                    return waitForTasksToFinish();
                }
                finally {
                    localExecutor.shutdownNow();
                }
            }
            // Use an externally provided executor
            else {
                futures.addAll(startTasks(executor, subtasks));
                return waitForTasksToFinish();
            }
        }
        catch (Exception e) {
            error.compareAndSet(null, e);
            LOG.warn("AllRowsReader terminated. " + e.getMessage(), e);
            cancel();
            
            throw error.get();
        }
    }
80360
public boolean start()
    throws Exception
  {
    // ServerNetwork clusterServer = getServerNetwork(_selfServer);
    
    if (_selfServer.port() >= 0) {
      // _hmuxProtocol = HmuxProtocol.create();
      _httpProtocol = new ProtocolHttp();
      _httpProtocolBartender = new ProtocolHttp();

      _httpProtocol.extensionProtocol(_protocolBartenderPublic);
      _httpProtocolBartender.extensionProtocol(_protocolBartender);
      
      ProtocolHttp protocolPublic = _httpProtocol;
      
      /*
      if (_serverConfig.getPortBartender() < 0) {
        protocolPublic = _httpProtocolBartender;
      }
      */
      
      Config config = _config;
      
      PortTcpBuilder portBuilder = new PortTcpBuilder(config);
      portBuilder.portName("server");
      //portBuilder.serverSocket(_serverConfig.getServerSocket());
      portBuilder.protocol(protocolPublic);
      portBuilder.ampManager(AmpSystem.currentManager());
      
      // XXX: config timing issues, see network/0330, 
      // NetworkServerConfig.getClusterIdleTime()
      
      _portPublic = new TcpPortBartender(portBuilder, _selfServer);
      portBuilder.portDefault(8080);
        
      if (config.get("bartender.port", int.class, 0) >= 0) {
        portBuilder = new PortTcpBuilder(config);
        portBuilder.portName("bartender");
        //portBuilder.serverSocket(_serverConfig.getSocketBartender());
        portBuilder.protocol(_httpProtocolBartender);
        //portBuilder.portDefault(_serverConfig.getPortBartender());
        portBuilder.ampManager(AmpSystem.currentManager());
          
        _portBartender = new TcpPortBartender(portBuilder, _selfServer);
      }
      
      /*
      _bartenderPortConfig.configure(_portPublic);
      _serverConfig.configurePort(_portPublic);
      
      if (_portBartender != null) {
        _bartenderPortConfig.configure(_portBartender);
        
        _serverConfig.configurePort(_portBartender);
      }
      */
    }
    
    startClusterPort();
    
    // validateHub(_selfServer.getPod());
    
    return true;
  }
1073662
public String pipelineWrite(String key, DataGenerator dataGenerator, int max_pipe_keys, int min_pipe_keys)
            throws Exception {
        // Create a random key between [0,MAX_PIPE_KEYS]
        int pipe_keys = randomGenerator.nextInt(max_pipe_keys);

        // Make sure that the number of keys in the pipeline are at least
        // MIN_PIPE_KEYS
        pipe_keys = Math.max(min_pipe_keys, pipe_keys);

        DynoJedisPipeline pipeline = this.jedisClient.get().pipelined();
        Map<String, Response<String>> responses = new HashMap<>();

        /**
         * writeSingle returns a single string, so we want to create a
         * StringBuilder to append all the keys in the form "key_n". This is
         * just used to return a single string
         */
        StringBuilder sb = new StringBuilder();
        // Iterate across the number of keys in the pipeline and set
        for (int n = 0; n < pipe_keys; ++n) {
            String nth_key = key + "_" + n;
            sb.append(nth_key);
            Response<String> resp = pipeline.set(key, key + dataGenerator.getRandomValue() + key);
            responses.put(nth_key, resp);
        }
        pipeline.sync();

        return sb.toString();
    }
762565
protected void connect() throws Exception {
        String username = this.properties.getUsername();
        String password = this.properties.getPassword();

        if (this.properties.isEncryptedPassword() || this.properties.isEncryptedUsername()) {
            String privateKey = getPrivateKey();
            if (this.properties.isEncryptedUsername()) {
                final String decryptedUsername = AESHelper.decrypt(this.properties.getProperty(ENCRYPTED_USERNAME), privateKey);
                if (StringUtils.isEmpty(decryptedUsername)) {
                    throw new DatabaseEngineException("The encrypted username could not be decrypted.");
                }
                username = decryptedUsername;

            }

            if (this.properties.isEncryptedPassword()) {
                final String decryptedPassword = AESHelper.decrypt(this.properties.getProperty(ENCRYPTED_PASSWORD), privateKey);
                if (StringUtils.isEmpty(decryptedPassword)) {
                    throw new DatabaseEngineException("The encrypted password could not be decrypted.");
                }
                password = decryptedPassword;
            }
        }

        final String jdbc = getFinalJdbcConnection(this.properties.getJdbc());

        this.conn = DriverManager.getConnection(
                jdbc,
                username,
                password);

        if (this.properties.isSchemaSet()) {
            setSchema(this.properties.getSchema());
        }

        this.currentSchema = Optional.ofNullable(getSchema())
            .orElseThrow(() -> new DatabaseEngineException("Could not get current schema"));

        setTransactionIsolation();

        logger.debug("Connection successful.");
    }
771570
public void commit() throws Exception {
        final Map<String, Object> params = new HashMap<String, Object>();

        final String applicationId = getPropertyAsString("applicationId");
        if (applicationId != null) params.put("applicationId", applicationId);

        final String name = getName();
        if (name != null) params.put("name", name);

        final String fallbackNumber = getFallbackNumber();
        if (fallbackNumber != null) params.put("fallbackNumber", fallbackNumber);

        final String uri = getUri();
        client.post(uri, params);

        final JSONObject object = toJSONObject(client.get(uri, null));
        updateProperties(object);
    }
410286
@Override
	public void runFetchLoop() throws Exception {
		// the map from broker to the thread that is connected to that broker
		final Map<Node, SimpleConsumerThread<T>> brokerToThread = new HashMap<>();

		// this holds possible the exceptions from the concurrent broker connection threads
		final ExceptionProxy errorHandler = new ExceptionProxy(Thread.currentThread());

		// the offset handler handles the communication with ZooKeeper, to commit externally visible offsets
		final ZookeeperOffsetHandler zookeeperOffsetHandler = new ZookeeperOffsetHandler(kafkaConfig);
		this.zookeeperOffsetHandler = zookeeperOffsetHandler;

		PeriodicOffsetCommitter periodicCommitter = null;
		try {

			// offsets in the state may still be placeholder sentinel values if we are starting fresh, or the
			// checkpoint / savepoint state we were restored with had not completely been replaced with actual offset
			// values yet; replace those with actual offsets, according to what the sentinel value represent.
			for (KafkaTopicPartitionState<TopicAndPartition> partition : subscribedPartitionStates()) {
				if (partition.getOffset() == KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET) {
					// this will be replaced by an actual offset in SimpleConsumerThread
					partition.setOffset(OffsetRequest.EarliestTime());
				} else if (partition.getOffset() == KafkaTopicPartitionStateSentinel.LATEST_OFFSET) {
					// this will be replaced by an actual offset in SimpleConsumerThread
					partition.setOffset(OffsetRequest.LatestTime());
				} else if (partition.getOffset() == KafkaTopicPartitionStateSentinel.GROUP_OFFSET) {
					Long committedOffset = zookeeperOffsetHandler.getCommittedOffset(partition.getKafkaTopicPartition());
					if (committedOffset != null) {
						// the committed offset in ZK represents the next record to process,
						// so we subtract it by 1 to correctly represent internal state
						partition.setOffset(committedOffset - 1);
					} else {
						// if we can't find an offset for a partition in ZK when using GROUP_OFFSETS,
						// we default to "auto.offset.reset" like the Kafka high-level consumer
						LOG.warn("No group offset can be found for partition {} in Zookeeper;" +
							" resetting starting offset to 'auto.offset.reset'", partition);

						partition.setOffset(invalidOffsetBehavior);
					}
				} else {
					// the partition already has a specific start offset and is ready to be consumed
				}
			}

			// start the periodic offset committer thread, if necessary
			if (autoCommitInterval > 0) {
				LOG.info("Starting periodic offset committer, with commit interval of {}ms", autoCommitInterval);

				periodicCommitter = new PeriodicOffsetCommitter(
						zookeeperOffsetHandler,
						subscribedPartitionStates(),
						errorHandler,
						autoCommitInterval);
				periodicCommitter.setName("Periodic Kafka partition offset committer");
				periodicCommitter.setDaemon(true);
				periodicCommitter.start();
			}

			// Main loop polling elements from the unassignedPartitions queue to the threads
			while (running) {
				// re-throw any exception from the concurrent fetcher threads
				errorHandler.checkAndThrowException();

				// wait for max 5 seconds trying to get partitions to assign
				// if threads shut down, this poll returns earlier, because the threads inject the
				// special marker into the queue
				List<KafkaTopicPartitionState<TopicAndPartition>> partitionsToAssign =
						unassignedPartitionsQueue.getBatchBlocking(5000);
				// note: if there are more markers, remove them all
				partitionsToAssign.removeIf(MARKER::equals);

				if (!partitionsToAssign.isEmpty()) {
					LOG.info("Assigning {} partitions to broker threads", partitionsToAssign.size());
					Map<Node, List<KafkaTopicPartitionState<TopicAndPartition>>> partitionsWithLeaders =
							findLeaderForPartitions(partitionsToAssign, kafkaConfig);

					// assign the partitions to the leaders (maybe start the threads)
					for (Map.Entry<Node, List<KafkaTopicPartitionState<TopicAndPartition>>> partitionsWithLeader :
							partitionsWithLeaders.entrySet()) {
						final Node leader = partitionsWithLeader.getKey();
						final List<KafkaTopicPartitionState<TopicAndPartition>> partitions = partitionsWithLeader.getValue();
						SimpleConsumerThread<T> brokerThread = brokerToThread.get(leader);

						if (!running) {
							break;
						}

						if (brokerThread == null || !brokerThread.getNewPartitionsQueue().isOpen()) {
							// start new thread
							brokerThread = createAndStartSimpleConsumerThread(partitions, leader, errorHandler);
							brokerToThread.put(leader, brokerThread);
						}
						else {
							// put elements into queue of thread
							ClosableBlockingQueue<KafkaTopicPartitionState<TopicAndPartition>> newPartitionsQueue =
									brokerThread.getNewPartitionsQueue();

							for (KafkaTopicPartitionState<TopicAndPartition> fp : partitions) {
								if (!newPartitionsQueue.addIfOpen(fp)) {
									// we were unable to add the partition to the broker's queue
									// the broker has closed in the meantime (the thread will shut down)
									// create a new thread for connecting to this broker
									List<KafkaTopicPartitionState<TopicAndPartition>> seedPartitions = new ArrayList<>();
									seedPartitions.add(fp);
									brokerThread = createAndStartSimpleConsumerThread(seedPartitions, leader, errorHandler);
									brokerToThread.put(leader, brokerThread);
									newPartitionsQueue = brokerThread.getNewPartitionsQueue(); // update queue for the subsequent partitions
								}
							}
						}
					}
				}
				else {
					// there were no partitions to assign. Check if any broker threads shut down.
					// we get into this section of the code, if either the poll timed out, or the
					// blocking poll was woken up by the marker element
					Iterator<SimpleConsumerThread<T>> bttIterator = brokerToThread.values().iterator();
					while (bttIterator.hasNext()) {
						SimpleConsumerThread<T> thread = bttIterator.next();
						if (!thread.getNewPartitionsQueue().isOpen()) {
							LOG.info("Removing stopped consumer thread {}", thread.getName());
							bttIterator.remove();
						}
					}
				}

				if (brokerToThread.size() == 0 && unassignedPartitionsQueue.isEmpty()) {
					if (unassignedPartitionsQueue.close()) {
						LOG.info("All consumer threads are finished, there are no more unassigned partitions. Stopping fetcher");
						break;
					}
					// we end up here if somebody added something to the queue in the meantime --> continue to poll queue again
				}
			}
		}
		catch (InterruptedException e) {
			// this may be thrown because an exception on one of the concurrent fetcher threads
			// woke this thread up. make sure we throw the root exception instead in that case
			errorHandler.checkAndThrowException();

			// no other root exception, throw the interrupted exception
			throw e;
		}
		finally {
			this.running = false;
			this.zookeeperOffsetHandler = null;

			// if we run a periodic committer thread, shut that down
			if (periodicCommitter != null) {
				periodicCommitter.shutdown();
			}

			// clear the interruption flag
			// this allows the joining on consumer threads (on best effort) to happen in
			// case the initial interrupt already
			Thread.interrupted();

			// make sure that in any case (completion, abort, error), all spawned threads are stopped
			try {
				int runningThreads;
				do {
					// check whether threads are alive and cancel them
					runningThreads = 0;
					Iterator<SimpleConsumerThread<T>> threads = brokerToThread.values().iterator();
					while (threads.hasNext()) {
						SimpleConsumerThread<?> t = threads.next();
						if (t.isAlive()) {
							t.cancel();
							runningThreads++;
						} else {
							threads.remove();
						}
					}

					// wait for the threads to finish, before issuing a cancel call again
					if (runningThreads > 0) {
						for (SimpleConsumerThread<?> t : brokerToThread.values()) {
							t.join(500 / runningThreads + 1);
						}
					}
				}
				while (runningThreads > 0);
			}
			catch (InterruptedException ignored) {
				// waiting for the thread shutdown apparently got interrupted
				// restore interrupted state and continue
				Thread.currentThread().interrupt();
			}
			catch (Throwable t) {
				// we catch all here to preserve the original exception
				LOG.error("Exception while shutting down consumer threads", t);
			}

			try {
				zookeeperOffsetHandler.close();
			}
			catch (Throwable t) {
				// we catch all here to preserve the original exception
				LOG.error("Exception while shutting down ZookeeperOffsetHandler", t);
			}
		}
	}
85512
public function moveAsRoot()
    {
        if ($this->getIsNewRecord()) {
            throw new Exception('The node should not be new record.');
        }

        if ($this->isRoot()) {
            throw new Exception('The node already is root node.');
        }

        $delta = 1 - $this->lft;
        $this
            ->objects()
            ->filter([
                'lft__gte' => $this->lft,
                'rgt__lte' => $this->rgt,
                'root' => $this->root,
            ])
            ->update([
                'lft' => new Expression('lft' . sprintf('%+d', $delta)),
                'rgt' => new Expression('rgt' . sprintf('%+d', $delta)),
                'level' => new Expression('level' . sprintf('%+d', 1 - $this->level)),
                'root' => $this->getMaxRoot(),
            ]);

        $this
            ->shiftLeftRight($this->rgt + 1, $this->lft - $this->rgt - 1);

        return $this;
    }
496514
public static void main(String[] args) throws Exception {
    Builder builder = Builder.newBuilder();

    /**
     * The processing graph consists of a supplier streamlet that emits
     * random integers between 1 and 100. From there, a series of transformers
     * is applied. At the end of the graph, the original value is ultimately
     * unchanged.
     */
    builder.newSource(() -> ThreadLocalRandom.current().nextInt(100))
        .transform(new DoNothingTransformer<>())
        .transform(new IncrementTransformer(10))
        .transform(new IncrementTransformer(-7))
        .transform(new DoNothingTransformer<>())
        .transform(new IncrementTransformer(-3))
        .log();

    Config config = Config.defaultConfig();

    // Fetches the topology name from the first command-line argument
    String topologyName = StreamletUtils.getTopologyName(args);

    // Finally, the processing graph and configuration are passed to the Runner, which converts
    // the graph into a Heron topology that can be run in a Heron cluster.
    new Runner().run(topologyName, config, builder);
  }
298914
@Override
    public void     start() throws Exception
    {
        if ( !state.compareAndSet(State.LATENT, State.STARTED) )
        {
            throw new IllegalStateException();
        }

        try
        {
            client.create().creatingParentContainersIfNeeded().forPath(queuePath);
        }
        catch ( KeeperException.NodeExistsException ignore )
        {
            // this is OK
        }
        if ( lockPath != null )
        {
            try
            {
                client.create().creatingParentContainersIfNeeded().forPath(lockPath);
            }
            catch ( KeeperException.NodeExistsException ignore )
            {
                // this is OK
            }
        }

        if ( !isProducerOnly || (maxItems != QueueBuilder.NOT_SET) )
        {
            childrenCache.start();
        }

        if ( !isProducerOnly )
        {
            service.submit
                (
                    new Callable<Object>()
                    {
                        @Override
                        public Object call()
                        {
                            runLoop();
                            return null;
                        }
                    }
                );
        }
    }
765107
public void run() throws Exception {
    if (config == null) {
      return;
    }

    SimpleRequest req = newRequest();

    InetSocketAddress socketAddress = (InetSocketAddress) config.address;

    if (!config.tls)
      system.log().info("Using plaintext gRPC HTTP/2 connections");

    GrpcClientSettings settings = Utils.createGrpcClientSettings(socketAddress, config.tls, system);

    List<BenchmarkServiceClient> clients = new ArrayList<BenchmarkServiceClient>(config.channels);
    for (int i = 0; i < config.channels; i++) {
      BenchmarkServiceClient client = BenchmarkServiceClient.create(settings, mat, system.dispatcher());
      clients.add(client);
    }

    // Do a warmup first. It's the same as the actual benchmark, except that
    // we ignore the statistics.
    warmup(req, clients);

    long startTime = System.nanoTime();
    long endTime = startTime + TimeUnit.SECONDS.toNanos(config.duration);
    List<Histogram> histograms = doBenchmark(req, clients, endTime);
    long elapsedTime = System.nanoTime() - startTime;

    Histogram merged = merge(histograms);

    printStats(merged, elapsedTime);
    if (config.histogramFile != null) {
      saveHistogram(merged, config.histogramFile);
    }
    shutdown(clients);
  }
463048
@Execute
    public void process() throws Exception {
        if (!concatOr(outCb == null, doReset)) {
            return;
        }
        checkNull(inRaster1);
        RenderedImage map1RI = inRaster1.getRenderedImage();
        RenderedImage map2RI = null;
        if (inRaster2 == null) {
            map2RI = map1RI;
        } else {
            map2RI = inRaster2.getRenderedImage();
        }

        outCb = new CoupledFieldsMoments().process(map1RI, map2RI, pBins, pFirst, pLast, pm, binmode);
    }
700721
protected function processKey($key, $value, array &$config)
        {
            if (strpos($key, $this->nestSeparator) !== false) {
                $pieces = explode($this->nestSeparator, $key, 2);

                if (!strlen($pieces[0]) || !strlen($pieces[1])) {
                    throw new Exception(sprintf('Invalid key "%s"', $key));
                } elseif (!isset($config[$pieces[0]])) {
                    if ($pieces[0] === '0' && !empty($config)) {
                        $config = array($pieces[0] => $config);
                    } else {
                        $config[$pieces[0]] = array();
                    }
                } elseif (!Arrays::isArray($config[$pieces[0]])) {
                    throw new Exception(sprintf(
                        'Cannot create sub-key for "%s", as key already exists', $pieces[0]
                    ));
                }

                $this->processKey($pieces[1], $value, $config[$pieces[0]]);
            } else {
                if ($key === '@include') {
                    if ($this->directory === null) {
                        throw new Exception('Cannot process @include statement for a string config');
                    }

                    $reader  = clone $this;
                    $include = $reader->fromFile($this->directory . '/' . $value);
                    $config  = array_replace_recursive($config, $include);
                } else {
                    $config[$key] = $value;
                }
            }
        }
883145
@Override
	public void invoke() throws Exception {

		if (LOG.isDebugEnabled()) {
			LOG.debug(formatLogString("Start task code."));
		}

		// whatever happens in this scope, make sure that the local strategies are cleaned up!
		// note that the initialization of the local strategies is in the try-finally block as well,
		// so that the thread that creates them catches its own errors that may happen in that process.
		// this is especially important, since there may be asynchronous closes (such as through canceling).
		try {
			// initialize the serializers (one per channel) of the record writers
			initOutputWriters(this.eventualOutputs);

			// initialize the remaining data structures on the input and trigger the local processing
			// the local processing includes building the dams / caches
			try {
				int numInputs = driver.getNumberOfInputs();
				int numBroadcastInputs = this.config.getNumBroadcastInputs();
				
				initInputsSerializersAndComparators(numInputs);
				initBroadcastInputsSerializers(numBroadcastInputs);
				
				// set the iterative status for inputs and broadcast inputs
				{
					List<Integer> iterativeInputs = new ArrayList<Integer>();
					
					for (int i = 0; i < numInputs; i++) {
						final int numberOfEventsUntilInterrupt = getTaskConfig().getNumberOfEventsUntilInterruptInIterativeGate(i);
			
						if (numberOfEventsUntilInterrupt < 0) {
							throw new IllegalArgumentException();
						}
						else if (numberOfEventsUntilInterrupt > 0) {
							this.inputReaders[i].setIterative(numberOfEventsUntilInterrupt);
							iterativeInputs.add(i);
				
							if (LOG.isDebugEnabled()) {
								LOG.debug(formatLogString("Input [" + i + "] reads in supersteps with [" +
										+ numberOfEventsUntilInterrupt + "] event(s) till next superstep."));
							}
						}
					}
					this.iterativeInputs = asArray(iterativeInputs);
				}
				
				{
					List<Integer> iterativeBcInputs = new ArrayList<Integer>();
					
					for (int i = 0; i < numBroadcastInputs; i++) {
						final int numberOfEventsUntilInterrupt = getTaskConfig().getNumberOfEventsUntilInterruptInIterativeBroadcastGate(i);
						
						if (numberOfEventsUntilInterrupt < 0) {
							throw new IllegalArgumentException();
						}
						else if (numberOfEventsUntilInterrupt > 0) {
							this.broadcastInputReaders[i].setIterative(numberOfEventsUntilInterrupt);
							iterativeBcInputs.add(i);
				
							if (LOG.isDebugEnabled()) {
								LOG.debug(formatLogString("Broadcast input [" + i + "] reads in supersteps with [" +
										+ numberOfEventsUntilInterrupt + "] event(s) till next superstep."));
							}
						}
					}
					this.iterativeBroadcastInputs = asArray(iterativeBcInputs);
				}
				
				initLocalStrategies(numInputs);
			} catch (Exception e) {
				throw new RuntimeException("Initializing the input processing failed" +
					e.getMessage() == null ? "." : ": " + e.getMessage(), e);
			}

			if (!this.running) {
				if (LOG.isDebugEnabled()) {
					LOG.debug(formatLogString("Task cancelled before task code was started."));
				}
				return;
			}

			// pre main-function initialization
			initialize();

			// read the broadcast variables
			for (int i = 0; i < this.config.getNumBroadcastInputs(); i++) {
				final String name = this.config.getBroadcastInputName(i);
				readAndSetBroadcastInput(i, name, this.runtimeUdfContext);
			}

			// the work goes here
			run();
		}
		finally {
			// clean up in any case!
			closeLocalStrategiesAndCaches();
		}

		if (this.running) {
			if (LOG.isDebugEnabled()) {
				LOG.debug(formatLogString("Finished task code."));
			}
		} else {
			if (LOG.isDebugEnabled()) {
				LOG.debug(formatLogString("Task code cancelled."));
			}
		}
	}