AvroMessageContentWrapper { byte[] wrapContent(byte[] message, String id, long timestamp, Schema schema, Map<String, String> externalMetadata) { GenericRecord genericRecord = bytesToRecord(message, schema); try { genericRecord.put(METADATA_MARKER, metadataMap(id, timestamp, externalMetadata)); return recordToBytes(genericRecord, schema); } catch (Exception e) { if (e instanceof AvroRuntimeException && e.getMessage().equals("Not a valid schema field: __metadata")) { throw new AvroInvalidMetadataException("Schema does not contain mandatory __metadata field for Hermes internal metadata. Please fix topic schema.", e); } throw new WrappingException("Could not wrap avro message", e); } } @Inject  AvroMessageContentWrapper(Clock clock);   }
@Test @SuppressWarnings("unchecked") public void shouldWrappedMessageContainsMetadata() { byte[] wrappedMessage = avroMessageContentWrapper.wrapContent(content, id, timestamp, avroUser.getSchema(), Collections.emptyMap()); GenericRecord messageWithMetadata = bytesToRecord(wrappedMessage, avroUser.getSchema()); Map<Utf8, Utf8> metadata = (Map<Utf8, Utf8>) messageWithMetadata.get(METADATA_MARKER); assertThat(metadata.get(METADATA_MESSAGE_ID_KEY).toString()).isEqualTo(id); assertThat(valueOf(metadata.get(METADATA_TIMESTAMP_KEY).toString())).isEqualTo(timestamp); assertThat(wrappedMessage).contains(content); }