code: @SuppressWarnings("unchecked")
      @Override
      public void onMessage(StreamMessage message)
      {
          Map<String, Object> values;
  
          try
          {
              values = this.converter.toMap(message);
          }
          catch (ConvertFailException ex)
          {
              String logFormat = "Fail convert to hbase record. Dispose received message. : Message={0}";
              logger.warn(MessageFormat.format(logFormat, message), ex);
              ack();
              return;
          }
  
          // rowidを設定する。
          HBaseData data = new HBaseData();
          HBaseRow row = new HBaseRow();
          row.setId(values.get("timestamp").toString() + "_" + values.get("source").toString());
  
          List<String> bodyList = (List<String>) values.get("body");
  
          for (int i = 0; i < bodyList.size(); i++)
          {
              CellDefine cellDefine = this.cellDefineList.get(i);
  
              HBaseCell registCell = new HBaseCell();
              registCell.setFamily(cellDefine.family);
              registCell.setQualifier(cellDefine.qualifier);
              registCell.setValue(bodyList.get(i));
              row.getCells().add(registCell);
          }
  
          data.getRows().add(row);
          insert(data);
  
          ack();
      }
docs: {@inheritDoc}