MessageContentWrapper { public UnwrappedMessageContent unwrapAvro(byte[] data, Topic topic, Integer schemaId, Integer schemaVersion) { for (AvroMessageContentUnwrapper unwrapper : avroMessageContentUnwrappers) { if (unwrapper.isApplicable(data, topic, schemaId, schemaVersion)) { AvroMessageContentUnwrapperResult result = unwrapper.unwrap(data, topic, schemaId, schemaVersion); if (result.getStatus() == SUCCESS) { return result.getContent(); } } } logger.error("All attempts to unwrap Avro message for topic {} with schema version {} failed", topic.getQualifiedName(), schemaVersion); throw new SchemaMissingException(topic); } @Inject  MessageContentWrapper(JsonMessageContentWrapper jsonMessageContentWrapper,
                                 AvroMessageContentWrapper avroMessageContentWrapper,
                                 AvroMessageSchemaIdAwareContentWrapper schemaIdAwareContentWrapper,
                                 AvroMessageHeaderSchemaVersionContentWrapper headerSchemaVersionContentWrapper,
                                 AvroMessageHeaderSchemaIdContentWrapper headerSchemaIdContentWrapper,
                                 AvroMessageAnySchemaVersionContentWrapper anySchemaVersionContentWrapper,
                                 AvroMessageSchemaVersionTruncationContentWrapper schemaVersionTruncationContentWrapper); UnwrappedMessageContent unwrapJson(byte[] data); UnwrappedMessageContent unwrapAvro(byte[] data, Topic topic, Integer schemaId, Integer schemaVersion); byte[] wrapAvro(byte[] data, String id, long timestamp, Topic topic, CompiledSchema<Schema> schema, Map<String, String> externalMetadata); byte[] wrapJson(byte[] data, String id, long timestamp, Map<String, String> externalMetadata);  }
@Test public void shouldThrowExceptionWhenMessageCouldNotBeUnwrappedByAnySchema() { Topic topic = createTopic(); byte[] doesNotMatchAnySchema = EMPTY_SCHEMA.getBytes(); when(rateLimiter.tryAcquireOnlineCheckPermit()).thenReturn(true); catchException(messageContentWrapper).unwrapAvro(doesNotMatchAnySchema, topic, NO_ID_IN_HEADER, NO_VERSION_IN_HEADER); assertThat(caughtException() instanceof SchemaMissingException).isTrue(); assertMetrics(0, 0, 1, 1, 0, 0, 0, 0, 1, 0, 0, 0); }