Quarkus Kafka Streams with confluent registry

[smeana]

Hello

Is any working example for a kafka streams aplication using schema registry and confluent serDes?

Regards

[quarkus-bot[bot]]

/cc @alesj, @cescoffier, @gunnarmorling, @ozangunalp, @rquinio

[alesj]

Any existing Apicurio schema registry Quarkus example will do, as long as you replace Serdes properly ...

This should be easier when we get this issues sorted out:

• Support Karapace as an alternative Kafka schema registry devservice #24350
• dev services for confluent schema registry #19277

[gunnarmorling]

Keep in mind only that the Confluent Avro Serdes may or may not work as-is in native mode; there may be the need for some tweaks like substitutations.

[alesj]

@gunnarmorling what is the issue? Which "forbidden" stuff is used?

[gunnarmorling]

Oh, I don't really know, it was more meant as a general note of caution. Perhaps it just works fine.

…

Message ID: <quarkusio/quarkus/repo-discussions/24488/comments/2421464@ github.com>

[cescoffier]

@gunnarmorling We already have the substitutions for Confluent Avro Serde.

[smeana]

Do you have any example with apicurio SerDe deserializing an avro serialized with ConfluentSerde?

[smeana]

Thanks @gunnarmorling @alesj
We are testing a migration from spring boot to quarkus.
We tested Apicurio registry and its serdes, but even enabling confluent id handler, (SerdeConfig.ENABLE_CONFLUENT_ID_HANDLER) we have magic byte issues on tools such as kafka ui, offset explorer etc... that we have already working.
If you have any solution for this please let me know.

So we keep using the confluent Serdes and the apicurio compatibility endpoint (/apis/ccompat/v6) it works fine, but we have to deserialize with confluent serdes inside quarkus Apps.
At the end we are in the same picture, just changing confluent Registry for apicurio one.

[alesj]

We tested Apicurio registry and its serdes, but even enabling confluent id handler (SerdeConfig.ENABLE_CONFLUENT_ID_HANDLER) we have magic byte issues on tools such as kafka ui, offset explorer etc... that we have already working.
If you have any solution for this please let me know.

This is more of a question for Apicurio registry GH issues ...

[alesj]

So we keep using the confluent Serdes and the apicurio compatibility endpoint (/apis/ccompat/v6) it works fine, but we have to deserialize with confluent serdes inside quarkus Apps.
At the end we are in the same picture, just changing confluent Registry for apicurio one.

Why is this a problem?
It's just a different Serde class at the end ...
Where you get full OSS functionality of the Apicurio registry.

[smeana]

The only problem I found it is that I am not able to specify from the properties the confluent schema registry url, having the following issue
org.apache.kafka.common.config.ConfigException: Missing required configuration "schema.registry.url" which has no default value.

tried:
kafka-streams.schema.registry.url=http://localhost:8085
kafka.schema.registry.url=http://localhost:8085
quarkus.kafka-streams.schema-registry-url=http://localhost:8085

To internally call kafkaConfiguration.put(attribute, value)

[alesj]

Afais, you need to check how Apicurio or Confluent serde's handle this particular property,
and feed the right key/value par to where you're configuring serde's usage.

e.g. for existing Quarkus Reactive Messaging it is: quarkus.mp.messaging.connector.smallrye-kafka.apicurio.registry.url

[smeana]

Yes but we are using Confluent Schema Registry in a kafka streams app not smallrye at all. None of the propertries tried works fine. It is needed to create the config like in vanilla java.

Thanks

[alesj]

I'm confused now, does it work or not? :-)

As I said, if you're not using smallrye or some other existing "magic",
then you're on your own to pick-up that proper config props and feed them to the right place and in the right shape.

[smeana]

Hello @alesj
In terms of the properties for schema Registry, it works creating the properties manually like in vanilla java (new Properties()...)
I couldn't manage to deserialize with apicurio serDes messages previously serialized with ConfluentSerDes
Thanks

[alesj]

I couldn't manage to deserialize with apicurio serDes messages previously serialized with ConfluentSerDes

Afair, this should work -- there was a test covering this ...
Did you check with Apicurio devs?

[carlesarnal]

Hi @smeana, when using apicurio serdes, in addition to using SerdeConfig.ENABLE_CONFLUENT_ID_HANDLER you must also configure this to use the contentId.

[carlesarnal]

And, for the question on how to mix the serdes, you can see an example on how to configure the apicurio serdes for that to work in our tests.

[lukens]

I don't suppose there are any better examples of this yet? I'm kind of completely lost here, and the comments in this thread haven't helped.

I've looked at the guide here on using a schema registry with Quarkus, but everything seems to work mostly by magic there, and I'm not sure how to translate that in a Kafka Streams Quarkus application.

I believe the key part from the first reply is "as long as you replace Serdes properly", and I'm not sure where to really start on that.

I'm trying to get this working with an already running Apicurio registry (rather than a dev container), and am currently trying to get it to work with a GenericRecord/Data Avro object (because figuring out the workflow for getting schemas from the registry and compiling to java objects is something I'm trying to ignore for this first stage). Ideally I'd like to access the Apicurio Registry in Confluent compatibility mode, as we have not yet decided what we will go with in production, so it would be good if we can switch easily.

Any help or pointers would be hugely appreciated.

[lukens]

I may have managed to figure it out, this seems to be working, though not sure if it's the "right" way to do it:

    @Produces
    public Topology buildTopology() {

        try (Serde<GenericRecord> valueSerde = new GenericAvroSerde()) {

            final Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url", SCHEMA_URL);
            valueSerde.configure(serdeConfig, false);

            final StreamsBuilder streamsBuilder = new StreamsBuilder();

            streamsBuilder
                .stream(TOPIC_NAME, Consumed.with(Void(), valueSerde))
                ... // other streaming stuff here
                    
            return streamsBuilder.build();

        }

    }

Also, in build.gradle I needed to add this repository:

    maven {
        url "https://packages.confluent.io/maven/"
    }

and these dependencies:

    implementation 'io.quarkus:quarkus-confluent-registry-avro'
    implementation 'io.confluent:kafka-streams-avro-serde:7.4.0'

(This example is currently only using a value serde as I don't yet have keys)