JacksonRecordMapper implements RecordMapper<String, T> { @Override public Record<T> processRecord(final Record<String> record) throws Exception { return new GenericRecord<>(record.getHeader(), mapper.readValue(record.getPayload().getBytes(), type)); } JacksonRecordMapper(Class<T> type);  JacksonRecordMapper(final ObjectMapper mapper, final Class<T> type); @Override Record<T> processRecord(final Record<String> record);  }
@Test public void testMapRecord() throws Exception { String jsonTweet = "{\"id\":1,\"user\":\"foo\",\"message\":\"Hello\"}"; when(record.getPayload()).thenReturn(jsonTweet); Record<Tweet> actual = mapper.processRecord(record); Tweet tweet = actual.getPayload(); assertThat(actual.getHeader()).isEqualTo(header); assertThat(tweet.getId()).isEqualTo(1); assertThat(tweet.getUser()).isEqualTo("foo"); assertThat(tweet.getMessage()).isEqualTo("Hello"); }
@Test public void testMapIncompleteRecord() throws Exception { String jsonTweet = "{\"id\":1,\"user\":\"foo\"}"; when(record.getPayload()).thenReturn(jsonTweet); Record<Tweet> actual = mapper.processRecord(record); Tweet tweet = actual.getPayload(); assertThat(actual.getHeader()).isEqualTo(header); assertThat(tweet.getId()).isEqualTo(1); assertThat(tweet.getUser()).isEqualTo("foo"); assertThat(tweet.getMessage()).isNull(); }
@Test public void testMapEmptyRecord() throws Exception { String jsonTweet = "{}"; when(record.getPayload()).thenReturn(jsonTweet); Record<Tweet> actual = mapper.processRecord(record); Tweet tweet = actual.getPayload(); assertThat(actual.getHeader()).isEqualTo(header); assertThat(tweet.getId()).isEqualTo(0); assertThat(tweet.getUser()).isNull(); assertThat(tweet.getMessage()).isNull(); }