JsonRecordMapper implements RecordMapper<String, P> { @Override public Record<P> processRecord(final Record<String> record) { P mappedObject = jsonb.fromJson(record.getPayload(), type); return new GenericRecord<>(record.getHeader(), mappedObject); } JsonRecordMapper(Class<P> type);  JsonRecordMapper(final Jsonb jsonb, final Class<P> type); @Override Record<P> processRecord(final Record<String> record);  }
@Test public void testMapRecord() { 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() { 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() { 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(); }