Called from ClusterSyncManagerLeaderListener
@throws Exception
970081
public String initializedGroupStatus() throws Exception {

        String status = null;

        if (clusteringEnabled) {
            initClusterNodeStatus();
            status = updateNodeStatus();
        }

        return status;
    }
519846
public void stop() throws Exception {
		LOG.info("Stop job leader service.");

		if (JobLeaderService.State.STARTED == state) {

			for (Tuple2<LeaderRetrievalService, JobLeaderService.JobManagerLeaderListener> leaderRetrievalServiceEntry: jobLeaderServices.values()) {
				LeaderRetrievalService leaderRetrievalService = leaderRetrievalServiceEntry.f0;
				JobLeaderService.JobManagerLeaderListener jobManagerLeaderListener = leaderRetrievalServiceEntry.f1;

				jobManagerLeaderListener.stop();
				leaderRetrievalService.stop();
			}

			jobLeaderServices.clear();
		}

		state = JobLeaderService.State.STOPPED;
	}
1260070
private void processParentEvent(WatchedEvent event) throws Exception {
        // get current children snapshot and reset this watch.
        Set<String> children = new TreeSet<String>(m_zk.getChildren(m_rootNode, m_parentWatch));
        // intersect to get newChildren and update m_lastChildren to the current set.
        Set<String> newChildren = new HashSet<String>(children);
        newChildren.removeAll(m_lastChildren);
        m_lastChildren = children;

        List<ByteArrayCallback> callbacks = new ArrayList<ByteArrayCallback>();
        for (String child : children) {
            ByteArrayCallback cb = new ByteArrayCallback();
            // set watches on new children.
            if(newChildren.contains(child)) {
                m_zk.getData(ZKUtil.joinZKPath(m_rootNode, child), m_childWatch, cb, null);
            } else {
                m_zk.getData(ZKUtil.joinZKPath(m_rootNode, child), false, cb, null);
            }

            callbacks.add(cb);
        }

        HashMap<String, JSONObject> cache = new HashMap<String, JSONObject>();
        for (ByteArrayCallback callback : callbacks) {
            try {
                byte payload[] = callback.get();
                JSONObject jsObj = new JSONObject(new String(payload, "UTF-8"));
                cache.put(callback.getPath(), jsObj);
            } catch (KeeperException.NoNodeException e) {
                // child may have been deleted between the parent trigger and getData.
            }
        }

        m_publicCache.set(ImmutableMap.copyOf(cache));
        if (m_cb != null) {
            m_cb.run(m_publicCache.get());
        }
    }
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;
                        }
                    }
                );
        }
    }
1200266
public void sync() throws Exception {
        try {
            JedisClusterPipeline pipeline = new JedisClusterPipeline(clusterInfoCache);
            execute(pipeline);
            pipeline.sync();
        } catch (JedisMovedDataException jre) {
            // if MOVED redirection occurred, rebuilds cluster's slot cache,
            // recommended by Redis cluster specification
            connectionHandler.renewSlotCache();
            if (maxAttempts > 0) {
                maxAttempts--;
                sync();
                return;
            }
            throw jre;
        } catch (Exception e) {
            throw e;
        }
    }
81985
@Override
	public void start(LeaderRetrievalListener listener) {
		checkNotNull(listener, "Listener must not be null.");

		synchronized (startStopLock) {
			checkState(!started, "StandaloneLeaderRetrievalService can only be started once.");
			started = true;

			// directly notify the listener, because we already know the leading JobManager's address
			listener.notifyLeaderAddress(leaderAddress, leaderId);
		}
	}
1191728
protected void doStop() throws Exception
    {
        MultiException mex=new MultiException();
        
        
        Thread thread= Thread.currentThread();
        ClassLoader lastContextLoader= thread.getContextClassLoader();
        
        try
        {
            // Context listeners
            if (_contextListeners != null)
            {
                if (_webAppHandler != null)
                {
                    ServletContextEvent event= new ServletContextEvent(getServletContext());
                    
                    for (int i= LazyList.size(_contextListeners); i-- > 0;)
                    {
                        try 
                        {
                            ((ServletContextListener)LazyList.get(_contextListeners, i)).contextDestroyed(event);
                        }
                        catch (Exception e)
                        {
                            mex.add(e);
                        }
                    }
                }
            }
            _contextListeners= null;
            
            // Stop the context
            try
            {
                super.doStop();
            }
            catch (Exception e)
            {
                mex.add(e);
            }
            
            // clean up
            clearSecurityConstraints();
            
            if (_webAppHandler != null)
                removeHandler(_webAppHandler);
            _webAppHandler= null;
            
            if (_errorPages != null)
                _errorPages.clear();
            _errorPages= null;
            
            _webApp=null;
            _webInf=null;
            
            _configurations=null;
            
        }
        finally
        {
            thread.setContextClassLoader(lastContextLoader);
        }
        
        if (mex!=null)
            mex.ifExceptionThrow();
    }
671099
@Setup(Level.Trial)
  public void setup() throws Exception {
    super.setup(clientExecutor,
        ExecutorType.DIRECT,
        MessageSize.SMALL,
        responseSize,
        FlowWindowSize.MEDIUM,
        ChannelType.NIO,
        maxConcurrentStreams,
        channelCount);
    callCounter = new AtomicLong();
    completed = new AtomicBoolean();
    record = new AtomicBoolean();
    latch =
        startFlowControlledStreamingCalls(maxConcurrentStreams, callCounter, record, completed, 1);
  }
358293
@Override
    public void stop() throws Exception {
        if (!this.started) {
            logger.warn(String.format("management=%s is already stopped", this.name));
            return;
        }

        // this.nettyClientOpsThread.setStarted(false);

        for (ManagementEventListener lstr : managementEventListeners) {
            try {
                lstr.onServiceStopped();
            } catch (Throwable ee) {
                logger.error("Exception while invoking onServiceStopped", ee);
            }
        }

        // We store the original state first
        this.store();

        // Stop all associations
        FastMap<String, Association> associationsTemp = this.associations;
        for (FastMap.Entry<String, Association> n = associationsTemp.head(), end = associationsTemp.tail(); (n = n.getNext()) != end;) {
            Association associationTemp = n.getValue();
            if (associationTemp.isStarted()) {
                ((NettyAssociationImpl) associationTemp).stop();
            }
        }

        FastList<Server> tempServers = servers;
        for (FastList.Node<Server> n = tempServers.head(), end = tempServers.tail(); (n = n.getNext()) != end;) {
            Server serverTemp = n.getValue();
            if (serverTemp.isStarted()) {
                try {
                    ((NettyServerImpl) serverTemp).stop();
                } catch (Exception e) {
                    logger.error(String.format("Exception while stopping the Server=%s", serverTemp.getName()), e);
                }
            }
        }

        // waiting till stopping associations
        for (int i1 = 0; i1 < 20; i1++) {
            boolean assConnected = false;
            for (FastMap.Entry<String, Association> n = this.associations.head(), end = this.associations.tail(); (n = n
                    .getNext()) != end;) {
                Association associationTemp = n.getValue();
                if (associationTemp.isConnected()) {
                    assConnected = true;
                    break;
                }
            }
            if (!assConnected)
                break;
            Thread.sleep(100);
        }

        // TODO - make a general shutdown and waiting for it instead of "waiting till stopping associations" 
        this.bossGroup.shutdownGracefully();
        this.workerGroup.shutdownGracefully();
        this.clientExecutor.shutdown();
       

        // TODO Should servers be also checked for shutdown?

        this.started = false;

    }
449246
void beginSynchronizing() throws IOException {
    // Synchronization is performed in other threads.
    Zxid lastZxid = persistence.getLatestZxid();
    ClusterConfiguration clusterConfig = persistence.getLastSeenConfig();
    long proposedEpoch = persistence.getProposedEpoch();
    for (PeerHandler ph : this.quorumMap.values()) {
      ph.setSyncTask(new SyncPeerTask(ph.getServerId(), ph.getLastZxid(),
                                      lastZxid, clusterConfig),
                     proposedEpoch);
      ph.startSynchronizingTask();
    }
  }
1352306
private void waitForHealthyCluster() throws Exception {
        Map<String, String> params = new HashMap<>();
        params.put("wait_for_status", this.clusterHealthColor);
        params.put("timeout", "30s");

        elasticSearchAdminClient.performRequest("GET", "/_cluster/health", params);
    }
1274295
private void postJoinGroupMembersError(final JoinGroupCompletionListener completionListener, final String errorMessage) {
        new Handler(Looper.getMainLooper()).post(new Runnable() {
            @Override
            public void run() {
                if (null != completionListener) {
                    completionListener.onError(errorMessage);
                }
            }
        });
    }
654424
@Override
  public void join(String peer) throws Exception {
    LOG.debug("Follower joins in.");
    try {
      LOG.debug("Query leader from {}", peer);
      Message query = MessageBuilder.buildQueryLeader();
      MessageTuple tuple = null;
      while (true) {
        // The joiner might gets started before the cluster gets started.
        // Instead of reporting failure immediately, the follower will retry
        // joining the cluster until joining the cluster successfully.
        try {
          sendMessage(peer, query);
          tuple = filter.getExpectedMessage(MessageType.QUERY_LEADER_REPLY,
                                            peer, config.getTimeoutMs());
          break;
        } catch (TimeoutException ex) {
          long retryInterval = 1;
          LOG.warn("Timeout while contacting leader, going to retry after {} "
              + "second.", retryInterval);
          // Waits for 1 second.
          Thread.sleep(retryInterval * 1000);
        }
      }
      this.electedLeader = tuple.getMessage().getReply().getLeader();
      LOG.debug("Got current leader {}", this.electedLeader);

      /* -- Synchronizing phase -- */
      while (true) {
        try {
          joinSynchronization();
          break;
        } catch (TimeoutException | BackToElectionException ex) {
          LOG.debug("Timeout({} ms) in synchronizing, retrying. Last zxid : {}",
                    getSyncTimeoutMs(),
                    persistence.getLatestZxid());
          transport.clear(electedLeader);
          clearMessageQueue();
        }
      }
      // See if it can be restored from the snapshot file.
      restoreFromSnapshot();
      // Delivers all transactions in log before entering broadcasting phase.
      deliverUndeliveredTxns();

      /* -- Broadcasting phase -- */
      // Initialize the vote for leader election.
      this.election.specifyLeader(this.electedLeader);
      changePhase(Phase.BROADCASTING);
      accepting();
    } catch (InterruptedException e) {
      LOG.debug("Participant is canceled by user.");
      throw e;
    } catch (TimeoutException e) {
      LOG.debug("Didn't hear message from {} for {} milliseconds. Going"
                + " back to leader election.",
                this.electedLeader,
                this.config.getTimeoutMs());
      if (persistence.getLastSeenConfig() == null) {
        throw new JoinFailure("Fails to join cluster.");
      }
    } catch (BackToElectionException e) {
      LOG.debug("Got GO_BACK message from queue, going back to electing.");
      if (persistence.getLastSeenConfig() == null) {
        throw new JoinFailure("Fails to join cluster.");
      }
    } catch (LeftCluster e) {
      LOG.debug("Exit running : {}", e.getMessage());
      throw e;
    } catch (Exception e) {
      LOG.error("Caught exception", e);
      throw e;
    } finally {
      changePhase(Phase.FINALIZING);
    }
  }
814548
@Override
    public void in() throws Exception {
        if (data == null) {
            throw new ComponentException("Not connected: " + toString());
        }
        Object val = data.getFieldValue();
        // fire only if there is a listener
        if (ens.shouldFire()) {
            DataflowEvent e = new DataflowEvent(ens.getController(), this, val);
////            DataflowEvent e = new DataflowEvent(ens.getController(), this, access.toObject());
            ens.fireIn(e);
//            // the value might be altered
            val = e.getValue();
        }

//        access.pass((Access) val);
        fa.setFieldValue(val);
    }
1068174
@Override
    protected void postInit() throws Exception {
        if (controlTree == null) {
            throw new RuntimeException("controlTree not set");
        }

        controlTree.create(NODES, false);
        controlTree.create(LEADERS, false);

        if (controlTree.exists(myNodeInfo.treeNodePath)) {
            LOG.error("A node with the name " + myNodeInfo.getName() + " already exists!");
            throw new RuntimeException("Initialization failure");
        }

        LOG.info("Required peer node properties: {}", requiredPeerNodeProperties);
        LOG.info("Required server properties: {}", requiredServerProperties);

        final Set<String> requiredProperties = (myId == 0 ? requiredServerProperties : requiredPeerNodeProperties);
        for (String property : requiredProperties) {
            if (!property.equals("id") && myNodeInfo.get(property) == null) {
                LOG.error("Required property {} not set!", property);
                throw new RuntimeException("Initialization failure");
            }
        }

        /// the calls in the demarcated section need to be in this specific order to avoid a possible race between nodes and leaders
        // `>>>> BEGIN CAREFULLY ORDERED SECTION
        branch = new DistributedBranchHelper(controlTree, NODES, false) {
            @Override
            protected boolean isNodeComplete(String node, Set<String> properties) {
                if (!properties.contains("id")) {
                    return false;
                }
                final short id = Short.parseShort(new String(controlTree.get(node + "/id"), Charsets.UTF_8));
                final Set<String> requiredProperties = (id == 0 ? requiredServerProperties : requiredPeerNodeProperties);
                final boolean success = properties.containsAll(requiredProperties);
                return success;
            }
        };

        branch.addListener(new DistributedTree.ListenerAdapter() {
            @Override
            public void nodeChildAdded(String parentPath, String childName) {
                AbstractCluster.this.nodeAdded(childName);
            }

            @Override
            public void nodeChildDeleted(String parentPath, String childName) {
                AbstractCluster.this.nodeRemoved(childName);
            }
        });
        // This read and handles the nodes.
        branch.init();

        controlTree.addListener(LEADERS, new DistributedTree.ListenerAdapter() {
            @Override
            public void nodeChildAdded(String parentPath, String childName) {
                AbstractCluster.this.leaderAdded(childName);
            }

            @Override
            public void nodeChildDeleted(String parentPath, String childName) {
                AbstractCluster.this.leaderRemoved(childName);
            }
        });

        myNodeInfo.writeToTree();

        setReady(true);

        super.postInit();

        joined = true;
        fireJoinedCluster();
        for (short id : masters.keySet()) {
            if (id != myId)
                fireNodeAdded(id);
        }
    }
1109521
private void doBind() throws Exception {
        LOG.debug("Creating listener socket");
        try {
            m_selector = Selector.open();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        ServerSocketChannel listenerSocket = ServerSocketChannel.open();
        InetSocketAddress inetsockaddr;
        if ((m_internalInterface == null) || (m_internalInterface.length() == 0)) {
            inetsockaddr = new InetSocketAddress(m_internalPort);
        }
        else {
            inetsockaddr = new InetSocketAddress(m_internalInterface, m_internalPort);
        }
        try {
            hostLog.info("Attempting to bind to internal ip " + inetsockaddr);
            listenerSocket.socket().bind(inetsockaddr);
            listenerSocket.configureBlocking(false);
            m_listenerSockets.add(listenerSocket);
        } catch (Exception e) {
            /*
             * If we bound to the leader address, the internal interface address  might not
             * bind if it is all interfaces
             */
            if (m_listenerSockets.isEmpty()) {
                LOG.fatal("Failed to bind to " + inetsockaddr);
                CoreUtils.printPortsInUse(hostLog);
                throw e;
            }
        }

        for (ServerSocketChannel ssc : m_listenerSockets) {
            ssc.register(m_selector, SelectionKey.OP_ACCEPT);
        }

        if (LOG.isDebugEnabled()) {
            LOG.debug("Non-Primary Listening on:" + inetsockaddr.toString());
        }
    }
899227
@Override
	public void invoke(IN next, Context context) throws Exception {
		// propagate asynchronous errors
		checkErroneous();

		byte[] serializedKey = schema.serializeKey(next);
		byte[] serializedValue = schema.serializeValue(next);
		String targetTopic = schema.getTargetTopic(next);
		if (targetTopic == null) {
			targetTopic = defaultTopicId;
		}

		int[] partitions = this.topicPartitionsMap.get(targetTopic);
		if (null == partitions) {
			partitions = getPartitionsByTopic(targetTopic, producer);
			this.topicPartitionsMap.put(targetTopic, partitions);
		}

		ProducerRecord<byte[], byte[]> record;
		if (flinkKafkaPartitioner == null) {
			record = new ProducerRecord<>(targetTopic, serializedKey, serializedValue);
		} else {
			record = new ProducerRecord<>(
					targetTopic,
					flinkKafkaPartitioner.partition(next, serializedKey, serializedValue, targetTopic, partitions),
					serializedKey,
					serializedValue);
		}
		if (flushOnCheckpoint) {
			synchronized (pendingRecordsLock) {
				pendingRecords++;
			}
		}
		producer.send(record, callback);
	}