code: private byte[] serialize() {
      if (messageBytes == null && message != null) {
        checkConverter();
        messageBytes = converter.toBytes(message);
        if (messageBytes == null) {
          // should we throw an IOException instead?
          LOG.warn("Could not serialize " + message.getClass());
        } else {
          message = null; // so that message and messageBytes don't go out of
                          // sync.
        }
      }
      return messageBytes;
    }
docs: Converts the message to raw bytes, and caches the converted value.
  @return converted value, which may be null in case of null message or error.