code: @Override
      public void buildTopology() throws Exception
      {
          // Get setting from StormConfig Object
          List<String> kestrelHosts = StormConfigUtil.getStringListValue(getConfig(),
                  KestrelJsonSpout.KESTREL_SERVERS);
          String kestrelQueueName = StormConfigUtil.getStringValue(getConfig(),
                  KestrelJsonSpout.KESTREL_QUEUE, "MessageQueue");
          int kestrelSpoutPara = StormConfigUtil.getIntValue(getConfig(), "kestrel.parallelism", 1);
          int printPara = StormConfigUtil.getIntValue(getConfig(), "print.parallelism", 1);
  
          // Topology Setting
          // Add Spout(KestrelJsonSpout)
          KestrelJsonSpout kestrelSpout = new KestrelJsonSpout(kestrelHosts, kestrelQueueName,
                  new StringScheme());
          getBuilder().setSpout("KestrelJsonSpout", kestrelSpout, kestrelSpoutPara);
  
          // Add Bolt(KestrelJsonSpout -> StreamMessagePrintBolt)
          StreamMessagePrintBolt streamMessagePrintBolt = new StreamMessagePrintBolt();
          getBuilder().setBolt("StreamMessagePrintBolt", streamMessagePrintBolt, printPara).fieldsGrouping(
                  "KestrelJsonSpout", new Fields(FieldName.MESSAGE_KEY));
  
          // Regist Serialize Setting.
          getConfig().registerSerialization(StreamMessage.class);
          getConfig().registerSerialization(StreamMessageHeader.class);
          getConfig().registerSerialization(MessageEntity.class);
      }
docs: {@inheritDoc}