code: public boolean bufferKestrelGet(int index)
      {
          KestrelClientInfo info = this.clientInfoList.get(index);
  
          long now = System.currentTimeMillis();
          if (now > info.blacklistTillTimeMs)
          {
              List<Item> items = null;
              try
              {
                  items = info.getValidClient().get(this.queueName, DEFAULT_BATCH_SIZE, 0,
                          this.messageTimeoutMs);
              }
              catch (TException e)
              {
                  blacklist(info, e);
                  return false;
              }
  
              Set<Long> toAck = Sets.newHashSet();
  
              for (Item item : items)
              {
                  Iterable<List<Object>> retItems = this.messageScheme.deserialize(item.get_data());
  
                  if (retItems != null)
                  {
                      for (List<Object> retItem : retItems)
                      {
                          EmitItem emitItem = generateEmitItem(retItem, new KestrelSourceId(index,
                                  item.get_id()));
  
                          if (!this.emitBuffer.offer(emitItem))
                          {
                              throw new RuntimeException(
                                      "KestrelThriftSpout's Internal Buffer Enqeueue Failed.");
                          }
                      }
  
                  }
                  else
                  {
                      toAck.add(item.get_id());
                  }
              }
  
              if (toAck.size() > 0)
              {
                  try
                  {
                      info.getClient().confirm(this.queueName, toAck);
                  }
                  catch (TException e)
                  {
                      blacklist(info, e);
                  }
              }
  
              if (items.size() > 0)
              {
                  return true;
              }
          }
          return false;
      }
docs: 指定したインデックスに対応したクライアント接続情報からメッセージを取得する。
  
  @param index クライアント接続情報インデックス
  @return 取得成功した場合true、失敗した場合false