2

Confluent SchemaRegistry 不适用于 Quarkus 和 MicroProfile

最终出现以下错误

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

我尝试构建一个 Quarkus kafka 消费者应用程序，该应用程序必须反序列化 kafka 主题中可用的 avro 序列化程序消息。当我尝试如下配置模式注册表（Confluent`s）以使用 avro 反序列化 kafka 消息时，我最终得到例外

配置

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

也尝试过

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();
    }
}

特性

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`

也尝试过

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
4

2 回答 2

1

给定格式

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

和属性schema.registry.url，你想要以下

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

kafkaConfiguration.put(attribute, value)在创建新消费者之前在内部调用

在任何情况下，JsonObject听起来都不是 Avro 子类，因此您还必须更改应用程序代码

于 2019-10-27T19:46:41.510 回答
0

对我来说，它通过在 kafka 键下设置模式注册表 url 来工作：

kafka.schema.registry.url=http://localhost:8081
于 2021-11-29T07:51:26.460 回答