Feature request: support schema repository

[bart-blommaerts]

Your library appears to one of the most mature libraries that supports the AWS Java SDK 2. Thank you for that.

Is it possible to support schema registries? AWS EventBridge doesn't work with SQS, so something like Spring Cloud Schema Registry would be a good match. You can wire it stand-alone in an application, but it will not do the schema transformation without Spring Cloud Stream. A simple example can be found here.

[JaidenAshmore]

Hey thanks for the interest!

I haven't seen this before so let me take a look at this and get back to you.

[JaidenAshmore]

Okay I have been mucking around with this today and I think I understand the request. Let me just write it all down and you can let me know where my thinking is wrong.

A bit of background
When writing this I didn't want it to be tied towards a specific framework, e.g. Spring/Spring Messaging, and so it doesn't do any logic to convert it to a Spring Messaging Message<?> type. The core library provides some basic argument resolvers, e.g. a MessageIdArgumentResolver, PayloadArgumentResolver etc but ultimately it allows for the consumers to easily extend the library to have any sort of ArgumentResolver that they desire.

This feature
So in this scenario it looks like we just need to implement our own ArgumentResolver that handles payloads that use Spring Cloud Schema Registry schemas for serialisation/deserialisation.

As an example let's say we receive a SQS message with the following information (I will use a JSON format that is definitely incorrectly formatted but just trying to show the basic functionality):

{
   "messageAttributes": [
          { "key": "contentType", "value": "application/prefix.sensor.v1+avro" }
    ],
   "body": "{serializedSensorRepresentation}"
}

We can create a listener that looks something like:

@QueueListener("myQueue")
public void processMessage(@AvroPayload Source source) {
        log.info("Got payload: {}", source);
}

and we would create our own custom ArgumentResolver that would look like:

@Component
public class AvroPayloadArgumentResolver implements ArgumentResolver<String> {
    private final SchemaRegistryClient schemaRegistryClient;
    private final AvroSchemaServiceManager avroSchemaServiceManager;

    @Autowired
    public AvroPayloadArgumentResolver (final SchemaRegistryClient schemaRegistryClient,
                                        final AvroSchemaServiceManager avroSchemaServiceManager) {
        this.schemaRegistryClient = schemaRegistryClient;
        this.avroSchemaServiceManager = avroSchemaServiceManager;
    }

    @Override
    public boolean canResolveParameter(final MethodParameter methodParameter) {
        return AnnotationUtils.findParameterAnnotation(methodParameter, AvroPayload.class)
                   .isPresent();
    }

    @Override
    public String resolveArgumentForParameter(final QueueProperties queueProperties,
                                              final MethodParameter methodParameter,
                                              final Message message) throws ArgumentResolutionException {
        final String contentType = message.messageAttributes().get("contentType").stringValue();

        final Pattern p = Pattern.compile("application/([^.]+)\\.([^.]+)\\.v([^.]+)+avro");
        final Matcher matcher = p.matcher(contentType);
        if (!matcher.matches()) {
            throw new ArgumentResolutionException("Could not extract content type");
        }

        final String subject = matcher.group(1);
        final String version = matcher.group(2);

        final Class<?> clazzToCastTo = methodParameter.getParameter().getType();

        final String schemaContent = schemaRegistryClient.fetch(new SchemaReference(subject, Integer.parseInt(version), AVRO_FORMAT));
        final Schema.Parser schemaParser = new Schema.Parser();
        final Schema writeSchema = schemaParser.parse(schemaContent);
        final Schema readSchema;
        try {
            readSchema = schemaParser.parse(RegistryArgumentResolver.class.getResourceAsStream("avro/sensor.avsc"));
        } catch (IOException e) {
            throw new RuntimeException("Could not parse schema", e);
        }

        try {
            return avroSchemaServiceManager.readData(clazzToCastTo, message.body().getBytes(), readSchema, writeSchema);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

That code above was just me mucking around with it, I haven't run it and I am surely doing something wrong with the Avro/Spring Cloud Schema Registry code but it should show the gist of it.

Does this seem to align with what you are thinking or am I missing something?

[bart-blommaerts]

Hi Jaiden,

First of all thanks for your time and for considering this feature.

Some background
I am currently building an event-driven system and my business stakeholders are new to "thinking in events". We also want to make it generic enough so it's not tied to any existing systems. Because of this, I'm confident the content of the events will change in the (foreseeable) future: fields need to be added, renamed, etc. Schema validation offers possibility to deal with backwards compatibily (ie. schema evolution).

Schema server
Spring Cloud Schema Registry offers a schema server: https://github.com/spring-cloud/spring-cloud-schema-registry/tree/master/spring-cloud-schema-registry-server that clients (event provider) can use to register their schema to: https://github.com/spring-cloud/spring-cloud-schema-registry/blob/master/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/DefaultSchemaRegistryClient.java. This is easy to set up, but is tied to Spring Cloud (it defaults to H2 for persistence, but JDBC is possible as well). The schema server is necessary, because there will be multiple services consuming multiple events.

Schema consumer
The DefaultSchemaRegistryClient can also fetch the schemas from the above registry using a SchemaReference. The server will contain multiple versions of the same schema (v1 / v2) and the consumer will need know how to deserialize both versions. In that regard, I don't think you need to schemaParser.parse twice. It should use the versioned schemaContent. You got the version from the message, so you only need to match that with the correct schema.

Ideally it would also return an AvroObject of the type of consumer schema instead of a String. This decouples the consumer schema from all provider schemas, but still offers backwards compatibility with older schemas (although I would argue the consumer needs to use the most recent producer version to keep it simple).

So, in summary, it looks like we're on the same page.

The overall setup is documented in:
https://github.com/spring-cloud/spring-cloud-stream/blob/master/docs/src/main/asciidoc/images/schema_reading.png

[JaidenAshmore]

Awesome, thanks for giving me some more info.

Overall it doesn't seem too difficult to do so I will try and find some time to flesh this out and put you on a PR when it is ready.

[JaidenAshmore]

Hey Bart,

I haven't gotten it working with the same sensor example that you linked me. Will be cleaning up the code and documentation and making sure I have added my necessary tests. Hopefully I will have something to show you by the end of the week unless something comes up before then.

[bart-blommaerts]

Thanks for the update. If you open a branch, I could join or have a look.

[JaidenAshmore]

Hey, the current branch I am working on is: https://github.com/JaidenAshmore/java-dynamic-sqs-listener/tree/140_spring_cloud_registry_extension

Once you build the app you can run the application locally by following the guide at: https://github.com/JaidenAshmore/java-dynamic-sqs-listener/tree/140_spring_cloud_registry_extension/examples/spring-cloud-schema-registry-example/README.md

You will see the following log lines as can be seen we are processing two versions of the schema.

18:00:23.881 [message-listener-message-processing-2] INFO  c.j.s.e.s.ConsumerApplication - Payload: {"id": "V1-a145c7ed-b125-4df2-a4f8-f0fb0cbc1ac5", "internalTemperature": 1.0, "externalTemperature": 0.0, "acceleration": 1.1, "velocity": 1.2}
18:00:24.455 [message-listener-message-processing-2] INFO  c.j.s.e.s.ConsumerApplication - Payload: {"id": "V2-f4610ebc-303e-4e19-9c1e-9f561e849e50", "internalTemperature": 1.0, "externalTemperature": 1.1, "acceleration": 1.2, "velocity": 1.3}
18:00:24.880 [message-listener-message-processing-2] INFO  c.j.s.e.s.ConsumerApplication - Payload: {"id": "V1-9602d6ea-1513-47f1-911f-96a939537610", "internalTemperature": 1.0, "externalTemperature": 0.0, "acceleration": 1.1, "velocity": 1.2}
18:00:25.883 [message-listener-message-processing-2] INFO  c.j.s.e.s.ConsumerApplication - Payload: {"id": "V1-3a9ebe05-a02a-4cb5-a0b0-29ee348735b0", "internalTemperature": 1.0, "externalTemperature": 0.0, "acceleration": 1.1, "velocity": 1.2}

Some tasks that I am needing to do are:

• Think about whether I have applied the right level of abstractions
• Cleanup edge cases etc
• Add unit/integration tests

Let me know if you have any suggestions on the current state. As I am new to Avro and the Spring Cloud Schema Registry I am sure I could be missing something.

Thanks!

[bart-blommaerts]

I tried the examples and they seem to working correctly! I submitted a very small PR for the examples.

I went through your commits and added some minor remarks.
I will try the update on a real project and provide you with feedback afterwards.

Have to say, I'm impressed so far!

[JaidenAshmore]

Nice, thanks so much for looking at it! Hopefully I can have a PR ready by the weekend and if all is well released by early next week

[bart-blommaerts]

Jayden,

some feedback / suggestions:

Receiving messages:

• When you use @EnableSchemaRegistrySqsExtension, maybe give an error / warning if no spring.cloud.schema-registry-client.endpoint has been set?
• For building, maybe add the possibility to skip the connection to the schema-registry-client? Or somehow make the user configurable. This links back to the use / configuration of EnableSchemaRegistrySqsExtension.
• software.amazon.awssdk is at 2.13.x
• When I receive a message in unspecified format, I get a com.jashmore.sqs.argument.ArgumentResolutionException: com.jashmore.sqs.argument.payload.mapper.PayloadMappingException: Error trying to resolve Payload for argument (which makes sense) and the message remains in flight. However, the consumer will receive the message again, causing an inifite loop. Maybe allow a backout to a DLQ (might have missed that) to prevent the infinite loop? Also this is deserialization exception, maybe it can be caught earlier, ie. when the schema is retrieved from the message.

Don't have any feedback so far wrt. publishing of messages.