code: @SuppressWarnings({"rawtypes"})
      @Override
      public void open(Map conf, TopologyContext context, SpoutOutputCollector collector)
      {
          super.open(conf, context, collector);
  
          this.ackWaitMap = new HashMap<>();
  
          int taskIndex = context.getThisTaskIndex();
          String targetUrl = this.brokerUrls.get(taskIndex);
          List<String> targetTopicStrs = this.subscribeTopics.get(taskIndex);
  
          MQTT mqtt = new MQTT();
          try
          {
              mqtt.setHost(targetUrl);
              this.connection = mqtt.blockingConnection();
              this.connection.connect();
  
              String connectMsgFormat = "MQTT Broker connected. : Url={0}";
              logger.info(MessageFormat.format(connectMsgFormat, targetUrl));
              List<Topic> topicList = Lists.newArrayList();
              for (String topicStr : targetTopicStrs)
              {
                  Topic topic = new Topic(topicStr, qos);
                  topicList.add(topic);
              }
              Topic[] targetTopics = topicList.toArray(new Topic[0]);
              this.connection.subscribe(targetTopics);
  
              String subscribeMsgFormat = "MQTT Broker subscribed. : Topic={0}";
              logger.info(MessageFormat.format(subscribeMsgFormat, Arrays.toString(targetTopics)));
          }
          catch (Exception ex)
          {
              String msg = "MQTT Broker connect failed. Skip initialize Spout.";
              logger.error(msg, ex);
          }
      }
docs: {@inheritDoc}