code: public static Message buildInvalidMessage(byte[] content) {
      InvalidMessage msg = InvalidMessage.newBuilder()
                           .setReceivedBytes(ByteString.copyFrom(content))
                           .build();
  
      return Message.newBuilder().setType(MessageType.INVALID_MESSAGE)
                                 .setInvalid(msg)
                                 .build();
    }
docs: Creates an INVALID message. This message will not be transmitted among
  workers. It's created when protobuf fails to parse the byte buffer.
  
  @param content the byte array which can't be parsed by protobuf.
  @return a message represents an invalid message.