code: @SuppressWarnings("rawtypes")
      @Override
      public void open(Map conf, TopologyContext context, SpoutOutputCollector collector)
      {
          Number timeout = (Number) conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS);
          this.messageTimeoutMs = (int) TimeUnit.SECONDS.toMillis(timeout.longValue());
          this.collector = collector;
          this.emitIndex = 0;
          this.clientInfoList = Lists.newArrayList();
          int numTasks = context.getComponentTasks(context.getThisComponentId()).size();
          int myIndex = context.getThisTaskIndex();
          int numHosts = this.hostInfos.size();
          if (numTasks < numHosts)
          {
              for (HostInfo host : this.hostInfos)
              {
                  this.clientInfoList.add(new KestrelClientInfo(host.getHost(), host.getPort()));
              }
          }
          else
          {
              HostInfo host = this.hostInfos.get(myIndex % numHosts);
              this.clientInfoList.add(new KestrelClientInfo(host.getHost(), host.getPort()));
          }
  
          this.emitBuffer = new LinkedList<EmitItem>();
      }
docs: {@inheritDoc}