/**
 * Copyright (C) 2015 DataTorrent, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.datatorrent.stram.engine;

import java.io.File;
import java.io.IOException;
import java.lang.Thread.State;
import java.lang.management.GarbageCollectorMXBean;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.util.AbstractMap.SimpleEntry;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

import net.engio.mbassy.bus.MBassador;
import net.engio.mbassy.bus.config.BusConfiguration;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.log4j.DTLoggerFactory;
import org.apache.log4j.LogManager;

import com.datatorrent.api.*;
import com.datatorrent.api.DAG.Locality;
import com.datatorrent.api.Operator.InputPort;
import com.datatorrent.api.Operator.OutputPort;
import com.datatorrent.api.Operator.ProcessingMode;
import com.datatorrent.api.StatsListener.OperatorRequest;
import com.datatorrent.api.annotation.Stateless;

import com.datatorrent.bufferserver.server.Server;
import com.datatorrent.bufferserver.storage.DiskStorage;
import com.datatorrent.bufferserver.util.Codec;
import com.datatorrent.common.util.ScheduledThreadPoolExecutor;
import com.datatorrent.netlet.DefaultEventLoop;
import com.datatorrent.netlet.util.Slice;
import com.datatorrent.stram.ComponentContextPair;
import com.datatorrent.stram.RecoverableRpcProxy;
import com.datatorrent.stram.StramUtils.YarnContainerMain;
import com.datatorrent.stram.StringCodecs;
import com.datatorrent.stram.api.*;
import com.datatorrent.stram.api.ContainerEvent.*;
import com.datatorrent.stram.api.OperatorDeployInfo.OperatorType;
import com.datatorrent.stram.api.OperatorDeployInfo.UnifierDeployInfo;
import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol.*;
import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol.OperatorHeartbeat.DeployState;
import com.datatorrent.stram.debug.StdOutErrLog;
import com.datatorrent.stram.plan.logical.LogicalPlan;
import com.datatorrent.stram.plan.logical.Operators.PortContextPair;
import com.datatorrent.stram.plan.logical.Operators.PortMappingDescriptor;
import com.datatorrent.stram.security.StramUserLogin;
import com.datatorrent.stram.stream.*;

/**
 * Object which controls the container process launched by {@link com.datatorrent.stram.StreamingAppMaster}.
 *
 * @since 0.3.2
 */
public class StreamingContainer extends YarnContainerMain
{
  public static final String PROP_APP_PATH = StreamingApplication.DT_PREFIX + Context.DAGContext.APPLICATION_PATH.getName();
  private final transient String jvmName;
  private final String containerId;
  private final transient StreamingContainerUmbilicalProtocol umbilical;
  protected final Map<Integer, Node<?>> nodes = new ConcurrentHashMap<Integer, Node<?>>();
  protected final Set<Integer> failedNodes = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
  private final Map<String, ComponentContextPair<Stream, StreamContext>> streams = new ConcurrentHashMap<String, ComponentContextPair<Stream, StreamContext>>();
  protected final Map<Integer, WindowGenerator> generators = new ConcurrentHashMap<Integer, WindowGenerator>();
  private transient List<GarbageCollectorMXBean> garbageCollectorMXBeans;
  /**
   * OIO groups map
   * key: operator id of oio owning thread node
   * value: list of nodes which are in oio with oio owning thread node
   */
  protected final Map<Integer, ArrayList<Integer>> oioGroups = new ConcurrentHashMap<Integer, ArrayList<Integer>>();
  private final Map<Stream, StreamContext> activeStreams = new ConcurrentHashMap<Stream, StreamContext>();
  private final Map<WindowGenerator, Object> activeGenerators = new ConcurrentHashMap<WindowGenerator, Object>();
  private int heartbeatIntervalMillis = 1000;
  private volatile boolean exitHeartbeatLoop = false;
  private final Object heartbeatTrigger = new Object();
  public static DefaultEventLoop eventloop;
  /**
   * List of listeners interested in listening into the status change of the nodes.
   */
  private long firstWindowMillis;
  private int windowWidthMillis;
  private InetSocketAddress bufferServerAddress;
  private com.datatorrent.bufferserver.server.Server bufferServer;
  private int checkpointWindowCount;
  private boolean fastPublisherSubscriber;
  private StreamingContainerContext containerContext;
  private List<StramToNodeRequest> nodeRequests;
  private final HashMap<String, Object> singletons;
  private final MBassador<ContainerEvent> eventBus; // event bus for publishing container events
  HashSet<Component<ContainerContext>> components;
  private RequestFactory requestFactory;

  static {
    try {
      eventloop = new DefaultEventLoop("ProcessWideEventLoop");
    }
    catch (IOException io) {
      throw new RuntimeException(io);
    }
  }

  protected StreamingContainer(String containerId, StreamingContainerUmbilicalProtocol umbilical)
  {
    this.jvmName = java.lang.management.ManagementFactory.getRuntimeMXBean().getName();
    this.components = new HashSet<Component<ContainerContext>>();
    this.eventBus = new MBassador<ContainerEvent>(BusConfiguration.Default(1, 1, 1));
    this.singletons = new HashMap<String, Object>();
    this.nodeRequests = new ArrayList<StramToNodeRequest>();

    logger.debug("instantiated StramChild {}", containerId);
    this.umbilical = umbilical;
    this.containerId = containerId;
  }

  @SuppressWarnings("unchecked")
  public void setup(StreamingContainerContext ctx)
  {
    containerContext = ctx;

    /* add a request factory local to this container */
    this.requestFactory = new RequestFactory();
    ctx.attributes.put(ContainerContext.REQUEST_FACTORY, requestFactory);

    heartbeatIntervalMillis = ctx.getValue(Context.DAGContext.HEARTBEAT_INTERVAL_MILLIS);
    firstWindowMillis = ctx.startWindowMillis;
    windowWidthMillis = ctx.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS);
    checkpointWindowCount = ctx.getValue(Context.DAGContext.CHECKPOINT_WINDOW_COUNT);

    fastPublisherSubscriber = ctx.getValue(LogicalPlan.FAST_PUBLISHER_SUBSCRIBER);

    Map<Class<?>, Class<? extends StringCodec<?>>> codecs = ctx.getValue(Context.DAGContext.STRING_CODECS);
    StringCodecs.loadConverters(codecs);

    try {
      if (ctx.deployBufferServer) {
        eventloop.start();

        int bufferServerRAM = ctx.getValue(ContainerContext.BUFFER_SERVER_MB);
        logger.debug("buffer server memory {}", bufferServerRAM);
        int blockCount;
        int blocksize;
        if (bufferServerRAM < ContainerContext.BUFFER_SERVER_MB.defaultValue) {
          blockCount = 8;
          blocksize = bufferServerRAM / blockCount;
          if (blocksize < 1) {
            blocksize = 1;
          }
        }
        else {
          blocksize = 64;
          blockCount = bufferServerRAM / blocksize;
        }
        // start buffer server, if it was not set externally
        bufferServer = new Server(0, blocksize * 1024 * 1024, blockCount);
        bufferServer.setAuthToken(ctx.getValue(StreamingContainerContext.BUFFER_SERVER_TOKEN));
        if (ctx.getValue(Context.DAGContext.BUFFER_SPOOLING)) {
          bufferServer.setSpoolStorage(new DiskStorage());
        }
        SocketAddress bindAddr = bufferServer.run(eventloop);
        logger.debug("Buffer server started: {}", bindAddr);
        this.bufferServerAddress = NetUtils.getConnectAddress(((InetSocketAddress) bindAddr));
      }
    }
    catch (IOException ex) {
      logger.warn("deploy request failed due to {}", ex);
      throw new IllegalStateException("Failed to deploy buffer server", ex);
    }

    for (Class<?> clazz : ContainerEvent.CONTAINER_EVENTS_LISTENERS) {
      try {
        Object newInstance = clazz.newInstance();
        singletons.put(clazz.getName(), newInstance);

        if (newInstance instanceof Component) {
          components.add((Component<ContainerContext>) newInstance);
        }

        eventBus.subscribe(newInstance);
      }
      catch (InstantiationException ex) {
        logger.warn("Container Event Listener Instantiation", ex);
      }
      catch (IllegalAccessException ex) {
        logger.warn("Container Event Listener Instantiation", ex);
      }
    }

    operateListeners(ctx, true);
  }

  public String getContainerId()
  {
    return this.containerId;
  }

  /**
   * This method is introduced as replacement for getTupleRecorder method which was cluttering the code.
   *
   * @param classname
   * @return
   */
  public Object getInstance(String classname)
  {
    return singletons.get(classname);
  }

  /**
   * Initialize container. Establishes heartbeat connection to the master
   * distribute through the callback address provided on the command line. Deploys
   * initial modules, then enters the heartbeat loop, which will only terminate
   * once container receives shutdown request from the master. On shutdown,
   * after exiting heartbeat loop, shutdown all modules and terminate
   * processing threads.
   *
   * @param args
   * @throws Throwable
   */
  public static void main(String[] args) throws Throwable
  {
    StdOutErrLog.tieSystemOutAndErrToLog();
    logger.debug("PID: " + System.getenv().get("JVM_PID"));
    logger.info("Child starting with classpath: {}", System.getProperty("java.class.path"));

    String appPath = System.getProperty(PROP_APP_PATH);
    if (appPath == null) {
      logger.error("{} not set in container environment.", PROP_APP_PATH);
      System.exit(1);
    }

    int exitStatus = 1; // interpreted as unrecoverable container failure

    RecoverableRpcProxy rpcProxy = new RecoverableRpcProxy(appPath, new Configuration());
    final StreamingContainerUmbilicalProtocol umbilical = rpcProxy.getProxy();
    final String childId = System.getProperty(StreamingApplication.DT_PREFIX + "cid");
    try {
      StreamingContainerContext ctx = umbilical.getInitContext(childId);
      StreamingContainer stramChild = new StreamingContainer(childId, umbilical);
      logger.debug("Container Context = {}", ctx);
      stramChild.setup(ctx);
      try {
        /* main thread enters heartbeat loop */
        stramChild.heartbeatLoop();
        exitStatus = 0;
      }
      finally {
        stramChild.teardown();
      }
    }
    catch (Error error) {
      logger.error("Fatal error in container!", error);
      /* Report back any failures, for diagnostic purposes */
      String msg = ExceptionUtils.getStackTrace(error);
      umbilical.reportError(childId, null, "FATAL: " + msg);
    }
    catch (Exception exception) {
      logger.error("Fatal exception in container!", exception);
      /* Report back any failures, for diagnostic purposes */
      String msg = ExceptionUtils.getStackTrace(exception);
      umbilical.reportError(childId, null, msg);
    }
    finally {
      rpcProxy.close();
      DefaultMetricsSystem.shutdown();
      logger.info("Exit status for container: {}", exitStatus);
      LogManager.shutdown();
    }

    if (exitStatus != 0) {
      System.exit(exitStatus);
    }
  }

  public synchronized void deactivate()
  {
    ArrayList<Thread> activeThreads = new ArrayList<Thread>();
    ArrayList<Integer> activeOperators = new ArrayList<Integer>();

    for (Map.Entry<Integer, Node<?>> e : nodes.entrySet()) {
      Thread t = e.getValue().context.getThread();
      if (t == null || !t.isAlive()) {
        disconnectNode(e.getKey());
      }
      else {
        activeThreads.add(t);
        activeOperators.add(e.getKey());
        e.getValue().shutdown();
      }
    }

    try {
      Iterator<Integer> iterator = activeOperators.iterator();
      for (Thread t : activeThreads) {
        t.join(1000);
        if (!t.getState().equals(State.TERMINATED)) {
          t.interrupt();
        }
        disconnectNode(iterator.next());
      }
    }
    catch (InterruptedException ex) {
      logger.warn("Aborting wait for operators to get deactivated!", ex);
    }

    for (WindowGenerator wg : activeGenerators.keySet()) {
      wg.deactivate();
    }
    activeGenerators.clear();

    for (Stream stream : activeStreams.keySet()) {
      stream.deactivate();
    }
    activeStreams.clear();
  }

  private void disconnectNode(int nodeid)
  {
    Node<?> node = nodes.get(nodeid);
    disconnectWindowGenerator(nodeid, node);

    PortMappingDescriptor portMappingDescriptor = node.getPortMappingDescriptor();
    Iterator<String> outputPorts = portMappingDescriptor.outputPorts.keySet().iterator();
    while (outputPorts.hasNext()) {
      String sourceIdentifier = String.valueOf(nodeid).concat(Component.CONCAT_SEPARATOR).concat(outputPorts.next());
      ComponentContextPair<Stream, StreamContext> pair = streams.remove(sourceIdentifier);
      if (pair != null) {
        if (activeStreams.remove(pair.component) != null) {
          pair.component.deactivate();
          eventBus.publish(new StreamDeactivationEvent(pair));
        }

        if (pair.component instanceof Stream.MultiSinkCapableStream) {
          String sinks = pair.context.getSinkId();
          if (sinks == null) {
            logger.error("mux sinks found connected at {} with sink id null", sourceIdentifier);
          }
          else {
            String[] split = sinks.split(MuxStream.MULTI_SINK_ID_CONCAT_SEPARATOR);
            for (int i = split.length; i-- > 0; ) {
              ComponentContextPair<Stream, StreamContext> spair = streams.remove(split[i]);
              if (spair == null) {
                logger.error("mux is missing the stream for sink {}", split[i]);
              }
              else {
                if (activeStreams.remove(spair.component) != null) {
                  spair.component.deactivate();
                  eventBus.publish(new StreamDeactivationEvent(spair));
                }

                spair.component.teardown();
              }
            }
          }
        }
        else {
          // it's either inline stream or it's bufferserver publisher.
        }

        pair.component.teardown();
      }
    }

    Iterator<String> inputPorts = portMappingDescriptor.inputPorts.keySet().iterator();
    while (inputPorts.hasNext()) {
      String sinkIdentifier = String.valueOf(nodeid).concat(Component.CONCAT_SEPARATOR).concat(inputPorts.next());
      ComponentContextPair<Stream, StreamContext> pair = streams.remove(sinkIdentifier);
      if (pair != null) {
        if (activeStreams.remove(pair.component) != null) {
          pair.component.deactivate();
          eventBus.publish(new StreamDeactivationEvent(pair));
        }

        pair.component.teardown();
        /**
         * we should also make sure that if this stream is connected to mux stream,
         * we deregister it from the mux stream to avoid clogged sink problem.
         */
        ComponentContextPair<Stream, StreamContext> sourcePair = streams.get(pair.context.getSourceId());
        if (sourcePair != null) {
          if (sourcePair == pair) {
            /* for some reason we had the stream stored against both source and sink identifiers */
            streams.remove(pair.context.getSourceId());
          }
          else {
            /* the stream was one of the many streams sourced by a muxstream */
            unregisterSinkFromMux(sourcePair, sinkIdentifier);
          }
        }
      }
    }
  }

  private boolean unregisterSinkFromMux(ComponentContextPair<Stream, StreamContext> muxpair, String sinkIdentifier)
  {
    String[] sinks = muxpair.context.getSinkId().split(MuxStream.MULTI_SINK_ID_CONCAT_SEPARATOR);
    boolean found = false;
    for (int i = sinks.length; i-- > 0; ) {
      if (sinks[i].equals(sinkIdentifier)) {
        sinks[i] = null;
        found = true;
        break;
      }
    }

    if (found) {
      ((Stream.MultiSinkCapableStream) muxpair.component).setSink(sinkIdentifier, null);

      if (sinks.length == 1) {
        muxpair.context.setSinkId(null);
        streams.remove(muxpair.context.getSourceId());
        if (activeStreams.remove(muxpair.component) != null) {
          muxpair.component.deactivate();
          eventBus.publish(new StreamDeactivationEvent(muxpair));
        }
        muxpair.component.teardown();
      }
      else {
        StringBuilder builder = new StringBuilder(muxpair.context.getSinkId().length() - MuxStream.MULTI_SINK_ID_CONCAT_SEPARATOR.length() - sinkIdentifier.length());

        found = false;
        for (int i = sinks.length; i-- > 0; ) {
          if (sinks[i] != null) {
            if (found) {
              builder.append(MuxStream.MULTI_SINK_ID_CONCAT_SEPARATOR).append(sinks[i]);
            }
            else {
              builder.append(sinks[i]);
              found = true;
            }
          }
        }

        muxpair.context.setSinkId(builder.toString());
      }
    }
    else {
      logger.error("{} was not connected to stream connected to {}", sinkIdentifier, muxpair.context.getSourceId());
    }

    return found;
  }

  private void disconnectWindowGenerator(int nodeid, Node<?> node)
  {
    WindowGenerator chosen1 = generators.remove(nodeid);
    if (chosen1 != null) {
      chosen1.releaseReservoir(Integer.toString(nodeid).concat(Component.CONCAT_SEPARATOR).concat(Node.INPUT));
      // should we send the closure of the port to the node?

      int count = 0;
      for (WindowGenerator wg : generators.values()) {
        if (chosen1 == wg) {
          count++;
        }
      }

      if (count == 0) {
        activeGenerators.remove(chosen1);
        chosen1.deactivate();
        chosen1.teardown();
      }
    }
  }

  private synchronized void undeploy(List<Integer> nodeList)
  {
    /**
     * make sure that all the operators which we are asked to undeploy are in this container.
     */
    HashMap<Integer, Node<?>> toUndeploy = new HashMap<Integer, Node<?>>();
    for (Integer operatorId : nodeList) {
      Node<?> node = nodes.get(operatorId);
      if (node == null) {
        throw new IllegalArgumentException("Node " + operatorId + " is not hosted in this container!");
      }
      else if (toUndeploy.containsKey(operatorId)) {
        throw new IllegalArgumentException("Node " + operatorId + " is requested to be undeployed more than once");
      }
      else {
        toUndeploy.put(operatorId, node);
      }
    }

    ArrayList<Thread> joinList = new ArrayList<Thread>();
    ArrayList<Integer> discoList = new ArrayList<Integer>();
    for (Integer operatorId : nodeList) {
      Thread t = nodes.get(operatorId).context.getThread();
      if (t == null || !t.isAlive()) {
        disconnectNode(operatorId);
      }
      else {
        joinList.add(t);
        discoList.add(operatorId);
        nodes.get(operatorId).shutdown();
      }
    }

    try {
      Iterator<Integer> iterator = discoList.iterator();
      for (Thread t : joinList) {
        t.join(1000);
        if (!t.getState().equals(State.TERMINATED)) {
          t.interrupt();
        }
        disconnectNode(iterator.next());
      }
      logger.info("Undeploy complete.");
    }
    catch (InterruptedException ex) {
      logger.warn("Aborting wait for operators to get deactivated!", ex);
    }

    for (Integer operatorId : nodeList) {
      nodes.remove(operatorId);
    }
  }

  public void teardown()
  {
    operateListeners(containerContext, false);

    deactivate();

    assert (streams.isEmpty());

    eventBus.shutdown();

    nodes.clear();

    HashSet<WindowGenerator> gens = new HashSet<WindowGenerator>();
    gens.addAll(generators.values());
    generators.clear();
    for (WindowGenerator wg : gens) {
      wg.teardown();
    }

    if (bufferServer != null) {
      eventloop.stop(bufferServer);
      eventloop.stop();
    }

    gens.clear();
  }

  public void triggerHeartbeat()
  {
    synchronized (heartbeatTrigger) {
      heartbeatTrigger.notifyAll();
    }
  }

  public void heartbeatLoop() throws Exception
  {
    umbilical.log(containerId, "[" + containerId + "] Entering heartbeat loop..");
    logger.debug("Entering heartbeat loop (interval is {} ms)", this.heartbeatIntervalMillis);
    final YarnConfiguration conf = new YarnConfiguration();
    long tokenLifeTime = (long)(containerContext.getValue(LogicalPlan.TOKEN_REFRESH_ANTICIPATORY_FACTOR) * containerContext.getValue(LogicalPlan.HDFS_TOKEN_LIFE_TIME));
    long expiryTime = System.currentTimeMillis();
    final Credentials credentials = UserGroupInformation.getCurrentUser().getCredentials();
    Iterator<Token<?>> iter = credentials.getAllTokens().iterator();
    while (iter.hasNext()) {
      Token<?> token = iter.next();
      logger.debug("token: {}", token);
    }
    String hdfsKeyTabFile = containerContext.getValue(LogicalPlan.KEY_TAB_FILE);
    while (!exitHeartbeatLoop) {

      if (UserGroupInformation.isSecurityEnabled() && System.currentTimeMillis() >= expiryTime && hdfsKeyTabFile != null) {
        expiryTime = StramUserLogin.refreshTokens(tokenLifeTime, "." + File.separator + "tmp", containerId, conf, hdfsKeyTabFile, credentials, null, false);
      }
      synchronized (this.heartbeatTrigger) {
        try {
          this.heartbeatTrigger.wait(heartbeatIntervalMillis);
        }
        catch (InterruptedException e1) {
          logger.warn("Interrupted in heartbeat loop, exiting..");
          break;
        }
      }

      long currentTime = System.currentTimeMillis();
      ContainerHeartbeat msg = new ContainerHeartbeat();
      msg.jvmName = jvmName;
      if (this.bufferServerAddress != null) {
        msg.bufferServerHost = this.bufferServerAddress.getHostName();
        msg.bufferServerPort = this.bufferServerAddress.getPort();
        if (bufferServer != null && !eventloop.isActive()) {
          logger.warn("Requesting restart due to terminated event loop");
          msg.restartRequested = true;
        }
      }
      msg.memoryMBFree = ((int) (Runtime.getRuntime().freeMemory() / (1024 * 1024)));
      garbageCollectorMXBeans = ManagementFactory.getGarbageCollectorMXBeans();
      for (GarbageCollectorMXBean bean : garbageCollectorMXBeans) {
        msg.gcCollectionTime += bean.getCollectionTime();
        msg.gcCollectionCount += bean.getCollectionCount();
      }

      ContainerHeartbeatResponse rsp;
      do {
        ContainerStats stats = new ContainerStats(containerId);
        // gather heartbeat info for all operators
        for (Map.Entry<Integer, Node<?>> e : nodes.entrySet()) {
          OperatorHeartbeat hb = new OperatorHeartbeat();
          hb.setNodeId(e.getKey());
          hb.setGeneratedTms(currentTime);
          hb.setIntervalMs(heartbeatIntervalMillis);
          if (e.getValue().commandResponse.size() > 0) {
            BlockingQueue<StatsListener.OperatorResponse> commandResponse = e.getValue().commandResponse;
            ArrayList<StatsListener.OperatorResponse> response = new ArrayList<StatsListener.OperatorResponse>();
            for (int i = 0; i < commandResponse.size(); i++) {
              response.add(commandResponse.poll());
            }
            hb.requestResponse = response;
          }
          OperatorContext context = e.getValue().context;
          context.drainStats(hb.getOperatorStatsContainer());

          if (context.getThread() == null || context.getThread().getState() != Thread.State.TERMINATED) {
            hb.setState(DeployState.ACTIVE);
          }
          else if (failedNodes.contains(hb.nodeId)) {
            hb.setState(DeployState.FAILED);
          }
          else {
            logger.debug("Reporting SHUTDOWN state because thread is {} and failedNodes is {}", context.getThread(), failedNodes);
            hb.setState(DeployState.SHUTDOWN);
          }

          stats.addNodeStats(hb);
        }

        /**
         * Container stats published for whoever is interested in listening.
         * Currently interested candidates are TupleRecorderCollection and BufferServerStatsSubscriber
         */
        eventBus.publish(new ContainerStatsEvent(stats));

        msg.setContainerStats(stats);

        // heartbeat call and follow-up processing
        //logger.debug("Sending heartbeat for {} operators.", msg.getContainerStats().size());
        msg.sentTms = System.currentTimeMillis();
        rsp = umbilical.processHeartbeat(msg);
        processHeartbeatResponse(rsp);
        if (rsp.hasPendingRequests) {
          logger.info("Waiting for pending request.");
          synchronized (this.heartbeatTrigger) {
            try {
              this.heartbeatTrigger.wait(500);
            }
            catch (InterruptedException ie) {
              logger.warn("Interrupted in heartbeat loop", ie);
              break;
            }
          }
        }
      }
      while (rsp.hasPendingRequests);

    }
    logger.debug("Exiting hearbeat loop");
    umbilical.log(containerId, "[" + containerId + "] Exiting heartbeat loop..");
  }

  private long lastCommittedWindowId = WindowGenerator.MIN_WINDOW_ID - 1;

  private void processNodeRequests(boolean flagInvalid)
  {
    for (StramToNodeRequest req : nodeRequests) {
      if (req.isDeleted()) {
        continue;
      }
      if (req instanceof StramToNodeChangeLoggersRequest) {
        handleChangeLoggersRequest((StramToNodeChangeLoggersRequest) req);
        continue;
      }

      Node<?> node = nodes.get(req.getOperatorId());
      if (node == null) {
        logger.warn("Node for operator {} is not found, probably not deployed yet", req.getOperatorId());
        continue;
      }

      Thread thread = node.context.getThread();
      if (thread == null || !thread.isAlive()) {
        if (flagInvalid) {
          logger.warn("Received request with invalid operator id {} ({})", req.getOperatorId(), req);
          req.setDeleted(true);
        }
      }
      else {
        logger.debug("request received: {}", req);
        OperatorRequest requestExecutor = requestFactory.getRequestExecutor(nodes.get(req.operatorId), req);
        if (requestExecutor != null) {
          node.context.request(requestExecutor);
        }
        else {
          logger.warn("No executor identified for the request {}", req);
        }
        req.setDeleted(true);
      }
    }
  }

  public void processHeartbeatResponse(ContainerHeartbeatResponse rsp)
  {
    if (rsp.nodeRequests != null) {
      nodeRequests = rsp.nodeRequests;
    }

    if (rsp.committedWindowId != lastCommittedWindowId) {
      lastCommittedWindowId = rsp.committedWindowId;
      OperatorRequest nr = null;
      for (Entry<Integer, Node<?>> e : nodes.entrySet()) {
        final Thread thread = e.getValue().context.getThread();
        if (thread == null || !thread.isAlive()) {
          continue;
        }

        if (e.getValue().getOperator() instanceof Operator.CheckpointListener) {
          if (nr == null) {
            nr = new OperatorRequest()
            {
              @Override
              public StatsListener.OperatorResponse execute(Operator operator, int operatorId, long windowId) throws IOException
              {
                ((Operator.CheckpointListener) operator).committed(lastCommittedWindowId);
                return null;
              }

            };
          }
          e.getValue().context.request(nr);
        }
      }
    }

    if (rsp.undeployRequest != null) {
      logger.info("Undeploy request: {}", rsp.undeployRequest);
      processNodeRequests(false);
      undeploy(rsp.undeployRequest);
    }

    if (rsp.shutdown) {
      logger.info("Received shutdown request");
      processNodeRequests(false);
      this.exitHeartbeatLoop = true;
      return;
    }

    if (rsp.deployRequest != null) {
      logger.info("Deploy request: {}", rsp.deployRequest);
      try {
        deploy(rsp.deployRequest);
      }
      catch (Exception e) {
        logger.error("deploy request failed", e);
        try {
          umbilical.log(this.containerId, "deploy request failed: " + rsp.deployRequest + " " + ExceptionUtils.getStackTrace(e));
        }
        catch (IOException ioe) {
          // ignore
        }
        this.exitHeartbeatLoop = true;
        throw new IllegalStateException("Deploy request failed: " + rsp.deployRequest, e);
      }
    }

    processNodeRequests(true);
  }

  private int getOutputQueueCapacity(List<OperatorDeployInfo> operatorList, int sourceOperatorId, String sourcePortName)
  {
    for (OperatorDeployInfo odi : operatorList) {
      if (odi.id == sourceOperatorId) {
        for (OperatorDeployInfo.OutputDeployInfo odiodi : odi.outputs) {
          if (odiodi.portName.equals(sourcePortName)) {
            return getValue(PortContext.QUEUE_CAPACITY, odiodi, odi);
          }
        }
      }
    }

    return PortContext.QUEUE_CAPACITY.defaultValue;
  }

  private synchronized void deploy(List<OperatorDeployInfo> nodeList) throws Exception
  {
    /*
     * A little bit of up front sanity check would reduce the percentage of deploy failures later.
     */
    for (OperatorDeployInfo ndi : nodeList) {
      if (nodes.containsKey(ndi.id)) {
        throw new IllegalStateException("Node with id: " + ndi.id + " already present in container " + containerId + "!");
      }
    }

    deployNodes(nodeList);

    HashMap<String, ArrayList<String>> groupedInputStreams = new HashMap<String, ArrayList<String>>();
    for (OperatorDeployInfo ndi : nodeList) {
      groupInputStreams(groupedInputStreams, ndi);
    }

    HashMap<String, ComponentContextPair<Stream, StreamContext>> newStreams = deployOutputStreams(nodeList, groupedInputStreams);
    deployInputStreams(nodeList, newStreams);
    for (ComponentContextPair<Stream, StreamContext> pair : newStreams.values()) {
      pair.component.setup(pair.context);
    }
    streams.putAll(newStreams);

    HashMap<Integer, OperatorDeployInfo> operatorMap = new HashMap<Integer, OperatorDeployInfo>(nodeList.size());
    for (OperatorDeployInfo o : nodeList) {
      operatorMap.put(o.id, o);
    }
    activate(operatorMap, newStreams);
  }

  public static String getUnifierInputPortName(String portName, int sourceNodeId, String sourcePortName)
  {
    return portName + "(" + sourceNodeId + Component.CONCAT_SEPARATOR + sourcePortName + ")";
  }

  private void massageUnifierDeployInfo(OperatorDeployInfo odi)
  {
    for (OperatorDeployInfo.InputDeployInfo idi : odi.inputs) {
      idi.portName = getUnifierInputPortName(idi.portName, idi.sourceNodeId, idi.sourcePortName);
    }
  }

  private void deployNodes(List<OperatorDeployInfo> nodeList) throws IOException
  {
    for (OperatorDeployInfo ndi : nodeList) {
      StorageAgent backupAgent = getValue(OperatorContext.STORAGE_AGENT, ndi);
      assert (backupAgent != null);

      Context parentContext;
      if (ndi instanceof UnifierDeployInfo) {
        OperatorContext unifiedOperatorContext = new OperatorContext(0, ((UnifierDeployInfo) ndi).operatorAttributes, containerContext);
        parentContext = new PortContext(ndi.inputs.get(0).contextAttributes, unifiedOperatorContext);
        massageUnifierDeployInfo(ndi);
      }
      else {
        parentContext = containerContext;
      }

      OperatorContext ctx = new OperatorContext(ndi.id, ndi.contextAttributes, parentContext);
      ctx.attributes.put(OperatorContext.ACTIVATION_WINDOW_ID, ndi.checkpoint.windowId);
      logger.debug("Restoring operator {} to checkpoint {} stateless={}.", ndi.id, Codec.getStringWindowId(ndi.checkpoint.windowId), ctx.stateless);
      Node<?> node = Node.retrieveNode(backupAgent.load(ndi.id, ctx.stateless ? Stateless.WINDOW_ID : ndi.checkpoint.windowId), ctx, ndi.type);
      node.currentWindowId = ndi.checkpoint.windowId;
      node.applicationWindowCount = ndi.checkpoint.applicationWindowCount;

      node.setId(ndi.id);
      nodes.put(ndi.id, node);
      logger.debug("Marking operator {} as deployed.", node);
    }
  }

  private HashMap.SimpleEntry<String, ComponentContextPair<Stream, StreamContext>> deployBufferServerPublisher(
    String connIdentifier, StreamCodec<?> streamCodec, long finishedWindowId, int queueCapacity, OperatorDeployInfo.OutputDeployInfo nodi)
    throws UnknownHostException
  {
    String sinkIdentifier = "tcp://".concat(nodi.bufferServerHost).concat(":").concat(String.valueOf(nodi.bufferServerPort)).concat("/").concat(connIdentifier);

    StreamContext bssc = new StreamContext(nodi.declaredStreamId);
    bssc.setPortId(nodi.portName);
    bssc.setSourceId(connIdentifier);
    bssc.setSinkId(sinkIdentifier);
    bssc.setFinishedWindowId(finishedWindowId);
    bssc.put(StreamContext.CODEC, streamCodec);
    bssc.put(StreamContext.EVENT_LOOP, eventloop);
    bssc.setBufferServerAddress(InetSocketAddress.createUnresolved(nodi.bufferServerHost, nodi.bufferServerPort));
    bssc.put(StreamContext.BUFFER_SERVER_TOKEN, nodi.bufferServerToken);
    InetAddress inetAddress = bssc.getBufferServerAddress().getAddress();
    if (inetAddress != null && NetUtils.isLocalAddress(inetAddress)) {
      bssc.setBufferServerAddress(new InetSocketAddress(InetAddress.getByName(null), nodi.bufferServerPort));
    }

    Stream publisher = fastPublisherSubscriber ? new FastPublisher(connIdentifier, queueCapacity * 256) : new BufferServerPublisher(connIdentifier, queueCapacity);
    return new HashMap.SimpleEntry<String, ComponentContextPair<Stream, StreamContext>>(sinkIdentifier, new ComponentContextPair<Stream, StreamContext>(publisher, bssc));
  }

  private HashMap<String, ComponentContextPair<Stream, StreamContext>> deployOutputStreams(
    List<OperatorDeployInfo> nodeList, HashMap<String, ArrayList<String>> groupedInputStreams)
    throws Exception
  {
    HashMap<String, ComponentContextPair<Stream, StreamContext>> newStreams = new HashMap<String, ComponentContextPair<Stream, StreamContext>>();
    /*
     * We proceed to deploy all the output streams. At the end of this block, our streams collection
     * will contain all the streams which originate at the output port of the operators. The streams
     * are generally mapped against the "nodename.portname" string. But the BufferServerPublishers which
     * share the output port with other inline streams are mapped against the Buffer Server port to
     * avoid collision and at the same time keep track of these buffer streams.
     */
    for (OperatorDeployInfo ndi : nodeList) {
      Node<?> node = nodes.get(ndi.id);
      long checkpointWindowId = ndi.checkpoint.windowId;

      for (OperatorDeployInfo.OutputDeployInfo nodi : ndi.outputs) {
        String sourceIdentifier = Integer.toString(ndi.id).concat(Component.CONCAT_SEPARATOR).concat(nodi.portName);
        int queueCapacity = getValue(PortContext.QUEUE_CAPACITY, nodi, ndi);
        logger.debug("for stream {} the queue capacity is {}", sourceIdentifier, queueCapacity);

        ArrayList<String> collection = groupedInputStreams.get(sourceIdentifier);
        Map<Integer, StreamCodec<?>> streamCodecs = nodi.streamCodecs;
        if ((collection == null) && (streamCodecs.size() == 1)) {
          assert (nodi.bufferServerHost != null) : "resulting stream cannot be inline: " + nodi;
          /*
           * Let's create a stream to carry the data to the Buffer Server.
           * Nobody in this container is interested in the output placed on this stream, but
           * this stream exists. That means someone outside of this container must be interested.
           */
          Map.Entry<Integer, StreamCodec<?>> entry = streamCodecs.entrySet().iterator().next();
          StreamCodec<?> streamCodec = entry.getValue();
          Integer streamCodecIdentifier = entry.getKey();
          String connIdentifier = sourceIdentifier + Component.CONCAT_SEPARATOR + streamCodecIdentifier;

          SimpleEntry<String, ComponentContextPair<Stream, StreamContext>> deployBufferServerPublisher =
            deployBufferServerPublisher(connIdentifier, streamCodec, checkpointWindowId, queueCapacity, nodi);
          newStreams.put(sourceIdentifier, deployBufferServerPublisher.getValue());
          node.connectOutputPort(nodi.portName, deployBufferServerPublisher.getValue().component);
        }
        else {
          /*
           * In this case we have 2 possibilities, either we have 1 inline or multiple streams.
           * Since we cannot tell at this point, we assume that we will have multiple streams and
           * plan accordingly. we possibly will come to this code block multiple times. We create
           * the MuxStream only the first time and use it for subsequent calls of this block.
           *
           * There is also the possibility that we have a stream with multiple sinks having distinct codecs
           */
          ComponentContextPair<Stream, StreamContext> pair = newStreams.get(sourceIdentifier);
          if (pair == null) {
            /**
             * Let's multiplex the output placed on this stream.
             * This container itself contains more than one parties interested.
             */
            StreamContext context = new StreamContext(nodi.declaredStreamId);
            context.setSourceId(sourceIdentifier);
            context.setFinishedWindowId(checkpointWindowId);
            Stream stream = new MuxStream();

            newStreams.put(sourceIdentifier, pair = new ComponentContextPair<Stream, StreamContext>(stream, context));
            node.connectOutputPort(nodi.portName, stream);
          }

          if (nodi.bufferServerHost != null) {
            /*
             * Although there is a node in this container interested in output placed on this stream, there
             * seems to at least one more party interested but placed in a container other than this one.
             */
            for (Map.Entry<Integer, StreamCodec<?>> entry : streamCodecs.entrySet()) {
              Integer streamCodecIdentifier = entry.getKey();
              StreamCodec<?> streamCodec = entry.getValue();

              String connIdentifier = sourceIdentifier + Component.CONCAT_SEPARATOR + streamCodecIdentifier;

              SimpleEntry<String, ComponentContextPair<Stream, StreamContext>> deployBufferServerPublisher =
                deployBufferServerPublisher(connIdentifier, streamCodec, checkpointWindowId, queueCapacity, nodi);
              newStreams.put(deployBufferServerPublisher.getKey(), deployBufferServerPublisher.getValue());

              String sinkIdentifier = pair.context.getSinkId();
              if (sinkIdentifier == null) {
                pair.context.setSinkId(deployBufferServerPublisher.getKey());
              }
              else {
                pair.context.setSinkId(sinkIdentifier.concat(", ").concat(deployBufferServerPublisher.getKey()));
              }

              ((Stream.MultiSinkCapableStream) pair.component).setSink(deployBufferServerPublisher.getKey(), deployBufferServerPublisher.getValue().component);
            }
          }
        }
      }
    }

    return newStreams;
  }

  /**
   * If the port is connected, return the declared stream Id.
   *
   * @param operatorId id of the operator to which the port belongs.
   * @param portname   name of port to which the stream is connected.
   * @return Stream Id if connected, null otherwise.
   */
  public final String getDeclaredStreamId(int operatorId, String portname)
  {
    String identifier = String.valueOf(operatorId).concat(Component.CONCAT_SEPARATOR).concat(portname);
    ComponentContextPair<Stream, StreamContext> spair = streams.get(identifier);
    if (spair == null) {
      return null;
    }

    return spair.context.getId();
  }

  @SuppressWarnings("unchecked")
  private void deployInputStreams(List<OperatorDeployInfo> operatorList, HashMap<String, ComponentContextPair<Stream, StreamContext>> newStreams) throws UnknownHostException
  {
    /*
     * collect any input operators along with their smallest window id,
     * those are subsequently used to setup the window generator
     */
    ArrayList<OperatorDeployInfo> inputNodes = new ArrayList<OperatorDeployInfo>();
    long smallestCheckpointedWindowId = Long.MAX_VALUE;
    //a simple map which maps the oio node to it's the node which owns the thread.
    Map<Integer, Integer> oioNodes = new ConcurrentHashMap<Integer, Integer>();

    /*
     * Hook up all the downstream ports. There are 2 places where we deal with more than 1
     * downstream ports. The first one follows immediately for WindowGenerator. The second
     * case is when source for the input port of some node in this container is in another
     * container. So we need to create the stream. We need to track this stream along with
     * other streams,and many such streams may exist, we hash them against buffer server
     * info as we did for outputs but throw in the sinkid in the mix as well.
     */
    for (OperatorDeployInfo ndi : operatorList) {
      if (ndi.inputs == null || ndi.inputs.isEmpty()) {
        /*
         * This has to be InputNode, so let's hook the WindowGenerator to it.
         * A node which does not take any input cannot exist in the DAG since it would be completely
         * unaware of the windows. So for that reason, AbstractInputNode allows Component.INPUT port.
         */
        inputNodes.add(ndi);
        /*
         * When we activate the window Generator, we plan to activate it only from required windowId.
         */
        ndi.checkpoint = getFinishedCheckpoint(ndi);
        if (ndi.checkpoint.windowId < smallestCheckpointedWindowId) {
          smallestCheckpointedWindowId = ndi.checkpoint.windowId;
        }
      }
      else {
        Node<?> node = nodes.get(ndi.id);

        for (OperatorDeployInfo.InputDeployInfo nidi : ndi.inputs) {
          if (nidi.streamCodecs.size() != 1) {
            throw new IllegalStateException("Only one input codec configuration should be present");
          }
          Map.Entry<Integer, StreamCodec<?>> entry = nidi.streamCodecs.entrySet().iterator().next();
          Integer streamCodecIdentifier = entry.getKey();
          StreamCodec<?> streamCodec = entry.getValue();
          String sourceIdentifier = Integer.toString(nidi.sourceNodeId).concat(Component.CONCAT_SEPARATOR).concat(nidi.sourcePortName);
          String sinkIdentifier = Integer.toString(ndi.id).concat(Component.CONCAT_SEPARATOR).concat(nidi.portName);

          int queueCapacity = getValue(PortContext.QUEUE_CAPACITY, nidi, ndi);

          Checkpoint checkpoint = getFinishedCheckpoint(ndi);
          ComponentContextPair<Stream, StreamContext> pair = streams.get(sourceIdentifier);
          if (pair == null) {
            pair = newStreams.get(sourceIdentifier);
          }

          if (pair == null) {
            /*
             * We connect to the buffer server for the input on this port.
             * We have already placed all the output streams for all the operators in this container.
             * Yet, there is no stream which can source this port so it has to come from the buffer
             * server, so let's make a connection to it.
             */
            assert (nidi.locality != Locality.CONTAINER_LOCAL && nidi.locality != Locality.THREAD_LOCAL);

            StreamContext context = new StreamContext(nidi.declaredStreamId);
            context.setBufferServerAddress(InetSocketAddress.createUnresolved(nidi.bufferServerHost, nidi.bufferServerPort));
            InetAddress inetAddress = context.getBufferServerAddress().getAddress();
            if (inetAddress != null && NetUtils.isLocalAddress(inetAddress)) {
              context.setBufferServerAddress(new InetSocketAddress(InetAddress.getByName(null), nidi.bufferServerPort));
            }
            context.put(StreamContext.BUFFER_SERVER_TOKEN, nidi.bufferServerToken);
            String connIdentifier = sourceIdentifier + Component.CONCAT_SEPARATOR + streamCodecIdentifier;
            context.setPortId(nidi.portName);
            context.put(StreamContext.CODEC, streamCodec);
            context.put(StreamContext.EVENT_LOOP, eventloop);
            context.setPartitions(nidi.partitionMask, nidi.partitionKeys);
            //context.setSourceId(sourceIdentifier);
            context.setSourceId(connIdentifier);
            context.setSinkId(sinkIdentifier);
            context.setFinishedWindowId(checkpoint.windowId);

            BufferServerSubscriber subscriber = fastPublisherSubscriber
              ? new FastSubscriber("tcp://".concat(nidi.bufferServerHost).concat(":").concat(String.valueOf(nidi.bufferServerPort)).concat("/").concat(connIdentifier), queueCapacity)
              : new BufferServerSubscriber("tcp://".concat(nidi.bufferServerHost).concat(":").concat(String.valueOf(nidi.bufferServerPort)).concat("/").concat(connIdentifier), queueCapacity);
            SweepableReservoir reservoir = subscriber.acquireReservoir(sinkIdentifier, queueCapacity);
            if (checkpoint.windowId >= 0) {
              node.connectInputPort(nidi.portName, new WindowIdActivatedReservoir(sinkIdentifier, reservoir, checkpoint.windowId));
            }
            node.connectInputPort(nidi.portName, reservoir);

            newStreams.put(sinkIdentifier, new ComponentContextPair<Stream, StreamContext>(subscriber, context));
            logger.debug("put input stream {} against key {}", subscriber, sinkIdentifier);
          }
          else {
            assert (nidi.locality == Locality.CONTAINER_LOCAL || nidi.locality == Locality.THREAD_LOCAL);
            /* we are still dealing with the MuxStream originating at the output of the source port */
            StreamContext inlineContext = new StreamContext(nidi.declaredStreamId);
            inlineContext.setSourceId(sourceIdentifier);
            inlineContext.setSinkId(sinkIdentifier);

            Stream stream;
            switch (nidi.locality) {
              case CONTAINER_LOCAL:
                int outputQueueCapacity = getOutputQueueCapacity(operatorList, nidi.sourceNodeId, nidi.sourcePortName);
                if (outputQueueCapacity > queueCapacity) {
                  queueCapacity = outputQueueCapacity;
                }

                stream = new InlineStream(queueCapacity);
                if (checkpoint.windowId >= 0) {
                  node.connectInputPort(nidi.portName, new WindowIdActivatedReservoir(sinkIdentifier, (SweepableReservoir) stream, checkpoint.windowId));
                }
                break;

              case THREAD_LOCAL:
                stream = new OiOStream();
                oioNodes.put(ndi.id, nidi.sourceNodeId);
                break;

              default:
                throw new IllegalStateException("Locality can be either ContainerLocal or ThreadLocal");
            }

            node.connectInputPort(nidi.portName, (SweepableReservoir) stream);
            newStreams.put(sinkIdentifier, new ComponentContextPair<Stream, StreamContext>(stream, inlineContext));

            if (!(pair.component instanceof Stream.MultiSinkCapableStream)) {
              String originalSinkId = pair.context.getSinkId();

              /* we come here only if we are trying to augment the dag */
              StreamContext muxContext = new StreamContext(nidi.declaredStreamId);
              muxContext.setSourceId(sourceIdentifier);
              muxContext.setFinishedWindowId(checkpoint.windowId);
              muxContext.setSinkId(originalSinkId);

              MuxStream muxStream = new MuxStream();
              muxStream.setSink(originalSinkId, pair.component);
              streams.put(originalSinkId, pair);

              Node<?> sourceNode = nodes.get(nidi.sourceNodeId);
              sourceNode.connectOutputPort(nidi.sourcePortName, muxStream);
              newStreams.put(sourceIdentifier, pair = new ComponentContextPair<Stream, StreamContext>(muxStream, muxContext));
            }

            /* here everything should be multisink capable */
            if (nidi.partitionKeys == null || nidi.partitionKeys.isEmpty()) {
              ((Stream.MultiSinkCapableStream) pair.component).setSink(sinkIdentifier, stream);
            }
            else {
              /*
               * generally speaking we do not have partitions on the inline streams so the control should not
               * come here but if it comes, then we are ready to handle it using the partition aware streams.
               */
              PartitionAwareSink<Object> pas = new PartitionAwareSink<Object>(streamCodec == null ? nonSerializingStreamCodec : (StreamCodec<Object>) streamCodec, nidi.partitionKeys, nidi.partitionMask, stream);
              ((Stream.MultiSinkCapableStream) pair.component).setSink(sinkIdentifier, pas);
            }

            String streamSinkId = pair.context.getSinkId();
            if (streamSinkId == null) {
              pair.context.setSinkId(sinkIdentifier);
            }
            else {
              pair.context.setSinkId(streamSinkId.concat(", ").concat(sinkIdentifier));
            }
          }
        }
      }
    }

    setupOiOGroups(oioNodes);

    if (!inputNodes.isEmpty()) {
      WindowGenerator windowGenerator = setupWindowGenerator(smallestCheckpointedWindowId);
      for (OperatorDeployInfo ndi : inputNodes) {
        generators.put(ndi.id, windowGenerator);

        Node<?> node = nodes.get(ndi.id);
        SweepableReservoir reservoir = windowGenerator.acquireReservoir(String.valueOf(ndi.id), 1024);
        if (ndi.checkpoint.windowId >= 0) {
          node.connectInputPort(Node.INPUT, new WindowIdActivatedReservoir(Integer.toString(ndi.id), reservoir, ndi.checkpoint.windowId));
        }
        node.connectInputPort(Node.INPUT, reservoir);
      }
    }

  }

  /**
   * Populates oioGroups with owner OIO Node as key and list of corresponding OIO nodes which will run in its thread as value
   * This method assumes that the DAG is valid as per OIO constraints
   */
  private void setupOiOGroups(Map<Integer, Integer> oioNodes)
  {
    for (Integer child : oioNodes.keySet()) {
      Integer oioParent = oioNodes.get(child);

      Integer temp;
      while ((temp = oioNodes.get(oioParent)) != null) {
        oioParent = temp;
      }

      ArrayList<Integer> children = oioGroups.get(oioParent);
      if (children == null) {
        oioGroups.put(oioParent, children = new ArrayList<Integer>());
      }
      children.add(child);
    }
  }

  /**
   * Create the window generator for the given start window id.
   * This is a hook for tests to control the window generation.
   *
   * @param finishedWindowId
   * @return WindowGenerator
   */
  protected WindowGenerator setupWindowGenerator(long finishedWindowId)
  {
    WindowGenerator windowGenerator = new WindowGenerator(new ScheduledThreadPoolExecutor(1, "WindowGenerator"), 1024);
    /**
     * let's make sure that we send the same window Ids with the same reset windows.
     */
    windowGenerator.setResetWindow(firstWindowMillis);

    long millisAtFirstWindow = WindowGenerator.getNextWindowMillis(finishedWindowId, firstWindowMillis, windowWidthMillis);
    windowGenerator.setFirstWindow(millisAtFirstWindow);
    windowGenerator.setWindowWidth(windowWidthMillis);

    long windowCount = WindowGenerator.getWindowCount(millisAtFirstWindow, firstWindowMillis, windowWidthMillis);
    windowGenerator.setCheckpointCount(checkpointWindowCount, (int) (windowCount % checkpointWindowCount));
    return windowGenerator;
  }

  private void setupNode(OperatorDeployInfo ndi)
  {
    failedNodes.remove(ndi.id);
    final Node<?> node = nodes.get(ndi.id);

    node.setup(node.context);

    /* setup context for all the input ports */
    LinkedHashMap<String, PortContextPair<InputPort<?>>> inputPorts = node.getPortMappingDescriptor().inputPorts;
    LinkedHashMap<String, PortContextPair<InputPort<?>>> newInputPorts = new LinkedHashMap<String, PortContextPair<InputPort<?>>>(inputPorts.size());
    for (OperatorDeployInfo.InputDeployInfo idi : ndi.inputs) {
      InputPort<?> port = inputPorts.get(idi.portName).component;
      PortContext context = new PortContext(idi.contextAttributes, node.context);
      newInputPorts.put(idi.portName, new PortContextPair<InputPort<?>>(port, context));
      port.setup(context);
    }
    inputPorts.putAll(newInputPorts);

    /* setup context for all the output ports */
    LinkedHashMap<String, PortContextPair<OutputPort<?>>> outputPorts = node.getPortMappingDescriptor().outputPorts;
    LinkedHashMap<String, PortContextPair<OutputPort<?>>> newOutputPorts = new LinkedHashMap<String, PortContextPair<OutputPort<?>>>(outputPorts.size());
    for (OperatorDeployInfo.OutputDeployInfo odi : ndi.outputs) {
      OutputPort<?> port = outputPorts.get(odi.portName).component;
      PortContext context = new PortContext(odi.contextAttributes, node.context);
      newOutputPorts.put(odi.portName, new PortContextPair<OutputPort<?>>(port, context));
      port.setup(context);
    }
    outputPorts.putAll(newOutputPorts);

    logger.debug("activating {} in container {}", node, containerId);
    /* This introduces need for synchronization on processNodeRequest which was solved by adding deleted field in StramToNodeRequest  */
    processNodeRequests(false);
    node.activate();
    eventBus.publish(new NodeActivationEvent(node));
  }

  private void teardownNode(OperatorDeployInfo ndi)
  {
    final Node<?> node = nodes.get(ndi.id);
    if (node == null) {
      logger.warn("node {}/{} took longer to exit, resulting in unclean undeploy!", ndi.id, ndi.name);
    }
    else {
      eventBus.publish(new NodeDeactivationEvent(node));
      node.deactivate();
      node.teardown();
      logger.debug("deactivated {}", node.getId());
    }
  }

  public synchronized void activate(final Map<Integer, OperatorDeployInfo> nodeMap, Map<String, ComponentContextPair<Stream, StreamContext>> newStreams)
  {
    for (ComponentContextPair<Stream, StreamContext> pair : newStreams.values()) {
      if (!(pair.component instanceof BufferServerSubscriber)) {
        activeStreams.put(pair.component, pair.context);
        pair.component.activate(pair.context);
        eventBus.publish(new StreamActivationEvent(pair));
      }
    }

    final CountDownLatch signal = new CountDownLatch(nodeMap.size());
    for (final OperatorDeployInfo ndi : nodeMap.values()) {
      /*
       * OiO nodes get activated with their primary nodes.
       */
      if (ndi.type == OperatorType.OIO) {
        continue;
      }

      final Node<?> node = nodes.get(ndi.id);
      new Thread(Integer.toString(ndi.id).concat("/").concat(ndi.name).concat(":").concat(node.getOperator().getClass().getSimpleName()))
      {
        @Override
        public void run()
        {
          HashSet<OperatorDeployInfo> setOperators = new HashSet<OperatorDeployInfo>();
          OperatorDeployInfo currentdi = ndi;
          try {
            /* primary operator initialization */
            setupNode(currentdi);
            setOperators.add(currentdi);

            /* lets go for OiO operator initialization */
            List<Integer> oioNodeIdList = oioGroups.get(ndi.id);
            if (oioNodeIdList != null) {
              for (Integer oioNodeId : oioNodeIdList) {
                currentdi = nodeMap.get(oioNodeId);
                setupNode(currentdi);
                setOperators.add(currentdi);
              }
            }

            currentdi = null;

            for (int i = setOperators.size(); i-- > 0; ) {
              signal.countDown();
            }

            node.run(); /* this is a blocking call */
          }
          catch (Error error) {
            int[] operators;
            if (currentdi == null) {
              logger.error("Voluntary container termination due to an error in operator set {}.", setOperators, error);
              operators = new int[setOperators.size()];
              int i = 0;
              for (Iterator<OperatorDeployInfo> it = setOperators.iterator(); it.hasNext(); i++) {
                operators[i] = it.next().id;
              }
            }
            else {
              logger.error("Voluntary container termination due to an error in operator {}.", currentdi, error);
              operators = new int[]{currentdi.id};
            }
            umbilical.reportError(containerId, operators, "Voluntary container termination due to an error. " + ExceptionUtils.getStackTrace(error));
            System.exit(1);
          }
          catch (Exception ex) {
            if (currentdi == null) {
              failedNodes.add(ndi.id);
              logger.error("Operator set {} stopped running due to an exception.", setOperators, ex);
              int[] operators = new int[]{ndi.id};
              umbilical.reportError(containerId, operators, "Stopped running due to an exception. " + ExceptionUtils.getStackTrace(ex));
            }
            else {
              failedNodes.add(currentdi.id);
              logger.error("Abandoning deployment of operator {} due to setup failure.", currentdi, ex);
              int[] operators = new int[]{currentdi.id};
              umbilical.reportError(containerId, operators, "Abandoning deployment due to setup failure. " + ExceptionUtils.getStackTrace(ex));
            }
          }
          finally {
            if (setOperators.contains(ndi)) {
              try {
                teardownNode(ndi);
              }
              catch (Exception ex) {
                failedNodes.add(ndi.id);
                logger.error("Shutdown of operator {} failed due to an exception.", ndi, ex);
              }
            }
            else {
              signal.countDown();
            }

            List<Integer> oioNodeIdList = oioGroups.get(ndi.id);
            if (oioNodeIdList != null) {
              for (Integer oioNodeId : oioNodeIdList) {
                OperatorDeployInfo oiodi = nodeMap.get(oioNodeId);
                if (setOperators.contains(oiodi)) {
                  try {
                    teardownNode(oiodi);
                  }
                  catch (Exception ex) {
                    failedNodes.add(oiodi.id);
                    logger.error("Shutdown of operator {} failed due to an exception.", oiodi, ex);
                  }
                }
                else {
                  signal.countDown();
                }
              }
            }
          }
        }

      }.start();
    }

    /**
     * we need to make sure that before any of the operators gets the first message, it's activated.
     */
    try {
      signal.await();
    }
    catch (InterruptedException ex) {
      logger.debug("Activation of operators interruped.", ex);
    }

    for (ComponentContextPair<Stream, StreamContext> pair : newStreams.values()) {
      if (pair.component instanceof BufferServerSubscriber) {
        activeStreams.put(pair.component, pair.context);
        pair.component.activate(pair.context);
        eventBus.publish(new StreamActivationEvent(pair));
      }
    }

    for (WindowGenerator wg : generators.values()) {
      if (!activeGenerators.containsKey(wg)) {
        activeGenerators.put(wg, generators);
        wg.activate(null);
      }
    }
  }

  private void groupInputStreams(HashMap<String, ArrayList<String>> groupedInputStreams, OperatorDeployInfo ndi)
  {
    for (OperatorDeployInfo.InputDeployInfo nidi : ndi.inputs) {
      String source = Integer.toString(nidi.sourceNodeId).concat(Component.CONCAT_SEPARATOR).concat(nidi.sourcePortName);

      /*
       * if we do not want to combine multiple streams with different partitions from the
       * same upstream node, we could also use the partition to group the streams together.
       * This logic comes with the danger that the performance of the group which shares
       * the same stream is bounded on the higher side by the performance of the lowest
       * performer upstream port. May be combining the streams is not such a good thing
       * but let's see if we allow this as an option to the user, what they end up choosing
       * the most.
       */
      ArrayList<String> collection = groupedInputStreams.get(source);
      if (collection == null) {
        collection = new ArrayList<String>();
        groupedInputStreams.put(source, collection);
      }
      collection.add(Integer.toString(ndi.id).concat(Component.CONCAT_SEPARATOR).concat(nidi.portName));
    }
  }

  protected Checkpoint getFinishedCheckpoint(OperatorDeployInfo ndi)
  {
    Checkpoint checkpoint;
    if (ndi.contextAttributes != null
      && ndi.contextAttributes.get(OperatorContext.PROCESSING_MODE) == ProcessingMode.AT_MOST_ONCE) {
      long now = System.currentTimeMillis();
      long windowCount = WindowGenerator.getWindowCount(now, firstWindowMillis, firstWindowMillis);

      Integer temp = ndi.contextAttributes.get(OperatorContext.APPLICATION_WINDOW_COUNT);
      if (temp == null) {
        temp = containerContext.getValue(OperatorContext.APPLICATION_WINDOW_COUNT);
      }
      int appWindowCount = (int) (windowCount % temp);

      temp = ndi.contextAttributes.get(OperatorContext.CHECKPOINT_WINDOW_COUNT);
      if (temp == null) {
        temp = containerContext.getValue(OperatorContext.CHECKPOINT_WINDOW_COUNT);
      }
      int lCheckpointWindowCount = (int) (windowCount % temp);
      checkpoint = new Checkpoint(WindowGenerator.getWindowId(now, firstWindowMillis, windowWidthMillis), appWindowCount, lCheckpointWindowCount);
      logger.debug("using {} on {} at {}", ProcessingMode.AT_MOST_ONCE, ndi.name, checkpoint);
    }
    else {
      checkpoint = ndi.checkpoint;
      logger.debug("using {} on {} at {}", ndi.contextAttributes == null ? ProcessingMode.AT_LEAST_ONCE :
        (ndi.contextAttributes.get(OperatorContext.PROCESSING_MODE) == null ? ProcessingMode.AT_LEAST_ONCE :
          ndi.contextAttributes.get(OperatorContext.PROCESSING_MODE)), ndi.name, checkpoint);
    }

    return checkpoint;
  }

  public void operateListeners(StreamingContainerContext ctx, boolean setup)
  {
    if (setup) {
      for (Component<ContainerContext> c : components) {
        c.setup(ctx);
      }
    }
    else {
      for (Component<ContainerContext> c : components) {
        c.teardown();
      }
    }
  }

  /**
   * Traverse the components to find out the default value set at the top level.
   * In OO this class should be shielded from using internal implementation. When
   * OperatorDeployInfo starts implementing proper semantics of Context, this
   * would not be needed.
   *
   * @param <T>         Type of the attribute.
   * @param key         Name of the attribute.
   * @param portContext Port context if applicable otherwise null.
   * @param deployInfo  Operator context if applicable otherwise null.
   * @return Value of the operator
   */
  private <T> T getValue(Attribute<T> key, com.datatorrent.api.Context.PortContext portContext, OperatorDeployInfo deployInfo)
  {
    if (portContext != null) {
      Attribute.AttributeMap attributes = portContext.getAttributes();
      if (attributes != null) {
        T attr = attributes.get(key);
        if (attr != null) {
          return attr;
        }
      }
    }

    if (deployInfo != null) {
      Attribute.AttributeMap attributes = deployInfo.contextAttributes;
      if (attributes != null) {
        T attr = attributes.get(key);
        if (attr != null) {
          return attr;
        }
      }
    }

    return containerContext.getValue(key);
  }

  private <T> T getValue(Attribute<T> key, OperatorDeployInfo deployInfo)
  {
    if (deployInfo != null) {
      Attribute.AttributeMap attributes = deployInfo.contextAttributes;
      if (attributes != null) {
        T attr = attributes.get(key);
        if (attr != null) {
          return attr;
        }
      }
    }

    return containerContext.getValue(key);
  }

  private void handleChangeLoggersRequest(StramToNodeChangeLoggersRequest request)
  {
    logger.debug("handle change logger request");
    DTLoggerFactory.getInstance().changeLoggersLevel(request.getTargetChanges());
  }

  private final StreamCodec<Object> nonSerializingStreamCodec = new StreamCodec<Object>()
  {
    @Override
    public Object fromByteArray(Slice fragment)
    {
      return null;
    }

    @Override
    public Slice toByteArray(Object o)
    {
      return null;
    }

    @Override
    public int getPartition(Object o)
    {
      return o.hashCode();
    }
  };

  private static final Logger logger = LoggerFactory.getLogger(StreamingContainer.class);
}