Do Quarkus & MicroProfile supports Confluent Schema Registry Avro Schema?

Viewed 1593

Confluent SchemaRegistry not working with Quarkus and MicroProfile

end up with the below error

***********ERROR    
Caused by: io.confluent.common.config.ConfigException: Missing required    configuration "schema.registry.url" which has no default value.

I try to build a Quarkus kafka consumer application which has to deserialize the avro serializer message available in a kafka topic.when I try to configure the Schema Registry (Confluent`s) as below to deserialize the kafka message using avro, I end up with the exception

Configuration

schema.registry.url=http://kafka-exposed:8081

kafka.bootstrap.servers=kafka-exposed:9200

# Configure the Kafka source (we read from it) 
mp.messaging.incoming.test-quarkustest.group.id=demo

mp.messaging.incoming.test-quarkustest.connector=smallrye-kafka

mp.messaging.incoming.test-quarkustest.topic=kafkacollectioncomplex

mp.messaging.incoming.test-quarkustest.value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer

mp.messaging.incoming.test-quarkustest.value.value.converter=io.confluent.connect.avro.AvroConverter

mp.messaging.incoming.test-
mp.messaging.incoming.test-quarkustest.value.converter.schemas.enable=true

TRIED AS BELOW AS WELL

mp.messaging.incoming.test-quarkustest.value.converter.schema.registry.url=http://kafka-exposed:8081
***********ERROR
Caused by: io.confluent.common.config.ConfigException: Missing required configuration "schema.registry.url" which has no default value.
@ApplicationScoped
public class ReadKafkaTopic {
    @Incoming("test-quarkustest")
    public CompletionStage<Void> process(KafkaMessage<String, JsonObject> message) {

        JsonObject data= message.getPayload();
        JsonArray array = (JsonArray) data.getJsonArray("skills");
        return message.ack();
    }
}

PROPERTIES

schema.registry.url=http://kafka-exposed:8081`
kafka.bootstrap.servers=kafka-exposed:9200`

# Configure the Kafka source (we read from it) 
mp.messaging.incoming.test-quarkustest.group.id=demo`
mp.messaging.incoming.test-quarkustest.connector=smallrye-kafka`
mp.messaging.incoming.test-quarkustest.topic=kafkacollectioncomplex`
mp.messaging.incoming.test-quarkustest.value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer`
mp.messaging.incoming.test-quarkustest.value.value.converter=io.confluent.connect.avro.AvroConverter`
mp.messaging.incoming.test-quarkustest.value.converter.schemas.enable=true`

TRIED AS BELOW AS WELL

mp.messaging.incoming.test-quarkustest.value.converter.schema.registry.url=http://kafka-exposed:8081
***********ERROR
Caused by: io.confluent.common.config.ConfigException: Missing required configuration "schema.registry.url" which has no default value.

ERROR***********************

2019-10-27 19:08:11,974 ERROR [io.sma.rea.mes.imp.ConfiguredChannelFactory] (main) Unable to create the publisher or subscriber during initialization: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer

76  at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:811)

77  at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:624)

78  at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:605)

79  at io.vertx.kafka.client.consumer.KafkaReadStream.create(KafkaReadStream.java:100)

80  at io.vertx.kafka.client.consumer.KafkaConsumer.create(KafkaConsumer.java:74)

81  at io.vertx.reactivex.kafka.client.consumer.KafkaConsumer.create(KafkaConsumer.java:168)

82  at io.smallrye.reactive.messaging.kafka.KafkaSource.<init>(KafkaSource.java:51)

83  at io.smallrye.reactive.messaging.kafka.KafkaConnector.getPublisherBuilder(KafkaConnector.java:65)

84  at io.smallrye.reactive.messaging.kafka.KafkaConnector_ClientProxy.getPublisherBuilder(KafkaConnector_ClientProxy.zig:203)

85  at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory.createPublisherBuilder(ConfiguredChannelFactory.java:145)

86  at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory.lambda$register$4(ConfiguredChannelFactory.java:123)

87  at java.util.HashMap.forEach(HashMap.java:1289)

88  at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory.register(ConfiguredChannelFactory.java:123)

89  at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory.initialize(ConfiguredChannelFactory.java:118)

90  at io.smallrye.reactive.messaging.impl.ConfiguredChannelFactory_ClientProxy.initialize(ConfiguredChannelFactory_ClientProxy.zig:195)

91  at java.util.Iterator.forEachRemaining(Iterator.java:116)

92  at java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801)

93  at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)

94  at io.smallrye.reactive.messaging.extension.MediatorManager.initializeAndRun(MediatorManager.java:132)

95  at io.smallrye.reactive.messaging.extension.MediatorManager_ClientProxy.initializeAndRun(MediatorManager_ClientProxy.zig:100)

96  at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart(SmallRyeReactiveMessagingLifecycle.java:20)

97  at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle_Observer_onApplicationStart_4e8937813d9e8faff65c3c07f88fa96615b70e70.notify(SmallRyeReactiveMessagingLifecycle_Observer_onApplicationStart_4e8937813d9e8faff65c3c07f88fa96615b70e70.zig:51)

98  at io.quarkus.arc.EventImpl$Notifier.notify(EventImpl.java:228)

99  at io.quarkus.arc.EventImpl.fire(EventImpl.java:69)

100     at io.quarkus.arc.runtime.LifecycleEventRunner.fireStartupEvent(LifecycleEventRunner.java:23)

101     at io.quarkus.arc.runtime.ArcRecorder.handleLifecycleEvents(ArcRecorder.java:103)

102     at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent32.deploy_0(LifecycleEventsBuildStep$startupEvent32.zig:77)

103     at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent32.deploy(LifecycleEventsBuildStep$startupEvent32.zig:36)

104     at io.quarkus.runner.ApplicationImpl1.doStart(ApplicationImpl1.zig:161)

105     at io.quarkus.runtime.Application.start(Application.java:94)

106     at io.quarkus.runtime.Application.run(Application.java:218)

107     at io.quarkus.runner.GeneratedMain.main(GeneratedMain.zig:41)

108 Caused by: io.confluent.common.config.ConfigException: Missing required configuration "schema.registry.url" which has no default value.

109     at io.confluent.common.config.ConfigDef.parse(ConfigDef.java:251)

110     at io.confluent.common.config.AbstractConfig.<init>(AbstractConfig.java:78)

111     at io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig.<init>(AbstractKafkaAvroSerDeConfig.java:105)

112     at io.confluent.kafka.serializers.KafkaAvroDeserializerConfig.<init>(KafkaAvroDeserializerConfig.java:41)

113     at io.confluent.kafka.serializers.KafkaAvroDeserializer.configure(KafkaAvroDeserializer.java:50)

114     at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:713)

115     ... 31 more
2 Answers

Given the format

mp.messaging.[outgoing|incoming].[channel-name].[attribute]=[value] 

And the attribute schema.registry.url, you'd want the following

mp.messaging.incoming.test-quarkustest.connector=smallrye-kafka
mp.messaging.incoming.test-quarkustest.connector=kafkacollectioncomplex
mp.messaging.incoming.test-quarkustest.bootstrap.servers=localhost:9092

# Setup Avro
mp.messaging.incoming.test-quarkustest.value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
mp.messaging.incoming.test-quarkustest.schema.registry.url=http://localhost:8081

Which internally will call kafkaConfiguration.put(attribute, value) before creating a new consumer


In any case, JsonObject doesn't sound an Avro subclass, so you'd have to change your application code as well

For me it works by setting the schema registry url under the kafka key:

kafka.schema.registry.url=http://localhost:8081
Related