JacksonRecordMarshaller implements RecordMarshaller<P, String> { @Override public JsonRecord processRecord(final Record<P> record) throws Exception { return new JsonRecord(record.getHeader(), mapper.writeValueAsString(record.getPayload())); } JacksonRecordMarshaller();  JacksonRecordMarshaller(final ObjectMapper mapper); @Override JsonRecord processRecord(final Record<P> record);  }
@Test public void marshal() throws Exception { Tweet tweet = new Tweet(1, "foo", "hi"); Record<Tweet> record = new GenericRecord<>(header, tweet); String expected = "{\"id\":1,\"user\":\"foo\",\"message\":\"hi\"}"; JsonRecord actual = marshaller.processRecord(record); assertThat(actual.getHeader()).isEqualTo(header); assertThat(actual.getPayload()).isEqualTo(expected); }