AvroMessageContentWrapper { @SuppressWarnings("unchecked") UnwrappedMessageContent unwrapContent(byte[] data, CompiledSchema<Schema> schema) { try { Map<Utf8, Utf8> metadata = (Map<Utf8, Utf8>) bytesToRecord(data, schema.getSchema()).get(METADATA_MARKER); MessageMetadata messageMetadata = getMetadata(metadata); return new UnwrappedMessageContent(messageMetadata, data, schema); } catch (Exception exception) { throw new UnwrappingException("Could not read avro message", exception); } } @Inject  AvroMessageContentWrapper(Clock clock);   }
@Test public void shouldUnwrapAvroMessageAndGenerateMetadataWhenNotExists() throws Throwable { byte [] wrappedMessage = wrapContentWithoutMetadata(content, avroUser.getSchema()); UnwrappedMessageContent unwrappedMessage = avroMessageContentWrapper.unwrapContent(wrappedMessage, avroUser.getCompiledSchema()); assertThat(unwrappedMessage.getMessageMetadata().getId()).isEmpty(); assertThat(unwrappedMessage.getMessageMetadata().getTimestamp()).isNotNull(); assertThat(unwrappedMessage.getContent()).startsWith(content); }
@Test public void shouldUnwrapAvroMessageAndSetEmptyMessageIdWhenNotGivenInMetadata() throws Throwable { byte [] wrappedMessage = wrapContentWithoutMessageIdInMetadata(content, avroUser.getSchema()); UnwrappedMessageContent unwrappedMessage = avroMessageContentWrapper.unwrapContent(wrappedMessage, avroUser.getCompiledSchema()); assertThat(unwrappedMessage.getMessageMetadata().getId()).isEmpty(); assertThat(unwrappedMessage.getMessageMetadata().getTimestamp()).isNotNull(); assertThat(unwrappedMessage.getContent()).contains(content); }