code: @Override
      @SuppressWarnings("unchecked")
      public void onMessage(StreamMessage message)
      {
          String endopointUri = getEndpointUri();
  
          Map<String, Object> resultMap;
  
          try
          {
              resultMap = this.converter.toMap(message);
          }
          catch (ConvertFailException ex)
          {
              String logFormat = "Fail convert to jdbc record. Dispose received message. : Message={0}";
              logger.warn(MessageFormat.format(logFormat, message), ex);
              ack();
              return;
          }
  
          List<Object> valueList = Lists.newArrayList();
  
          valueList.add(resultMap.get("messageId"));
          valueList.add(resultMap.get("timestamp"));
          valueList.add(resultMap.get("source"));
  
          List<String> bodyList = (List<String>) resultMap.get("body");
  
          valueList.addAll(bodyList);
  
          insert(endopointUri, valueList);
          ack();
      }
docs: Message受信時の処理を行う
  
  @param message 受信Message