### Question:
CloudEventMessageSerializer implements Serializer<MessageReader> { @Override public byte[] serialize(String topic, MessageReader data) { throw new UnsupportedOperationException("CloudEventMessageSerializer supports only the signature serialize(String, Headers, Message)"); }  @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] serialize(String topic, MessageReader data); @Override byte[] serialize(String topic, Headers headers, MessageReader data);  }### Answer:
@Test public void serializerShouldWork() { String topic = "test"; CloudEvent event = Data.V1_WITH_JSON_DATA; CloudEventMessageSerializer serializer = new CloudEventMessageSerializer(); Headers headers = new RecordHeaders(); MockBinaryMessageWriter inMessage = new MockBinaryMessageWriter(); CloudEventUtils.toVisitable(event).read(inMessage); byte[] payload = serializer.serialize(topic, headers, inMessage); MessageReader outMessage = KafkaMessageFactory.createReader(headers, payload); assertThat(outMessage.getEncoding()) .isEqualTo(Encoding.BINARY); assertThat(outMessage.toEvent()) .isEqualTo(event); }