Kotlin app configuration kafka consumer and produser with avro has failed
При попытке сконфигурировать кафку через application.yaml получаю ошибки при конфигурации consumer и producer вместе с авро сериализатором и дисериализатором. Приложение написано на kotlin. application.yaml:
spring:
application:
name: test
cloud:
consul:
enabled: false
stream:
bindings:
input:
destination: dest
group: group
consumer:
concurrency: 12
instanceCount: 2
partitioned: true
useNativeDecoding: true
output:
destination: dest
group: grout
producer:
partitionKeyExpression: headers['header']
partitionCount: 24
headerMode: headers
useNativeEncoding: true
kafka:
bindings:
input:
consumer:
startOffset: latest
resetOffsets: true
configuration:
client.id: test
key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
schema.registry.url: urls..
output:
producer:
configuration:
client.id: test
key.serializer: org.apache.kafka.common.serialization.StringSerializer
value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url: urls..
binder:
brokers: brokers..
configuration:
sasl.mechanism: PLAIN
security.protocol: SASL_PLAINTEXT
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";
required-acks: -1
replication-factor: 3
Ошибки:
Producer
org.apache.
kafka.common.KafkaException: Failed to construct kafka producer
at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:457)
at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:289)
at org.springframework.kafka.core.DefaultKafkaProducerFactory.createKafkaProducer(DefaultKafkaProducerFactory.java:318)
at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:305)
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.lambda$createProducerMessageHandler$0(KafkaMessageChannelBinder.java:287)
at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.lambda$getPartitionsForTopic$2(KafkaTopicProvisioner.java:410)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:164)
at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.getPartitionsForTopic(KafkaTopicProvisioner.java:405)
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createProducerMessageHandler(KafkaMessageChannelBinder.java:284)
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createProducerMessageHandler(KafkaMessageChannelBinder.java:132)
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:184)
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:90)
at org.springframework.cloud.stream.binder.AbstractBinder.bindProducer(AbstractBinder.java:151)
at org.springframework.cloud.stream.binding.BindingService.lambda$rescheduleProducerBinding$2(BindingService.java:290)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.common.KafkaException: Could not instantiate class io.confluent.kafka.serializers.KafkaAvroSerializer
at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:310)
at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:302)
at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:370)
... 22 common frames omitted
Caused by: java.lang.reflect.InvocationTargetException: null
at jdk.internal.reflect.GeneratedConstructorAccessor58.newInstance(Unknown Source)
at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:490)
at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:306)
... 24 common frames omitted
Caused by: java.lang.NoClassDefFoundError: io/confluent/common/Configurable
at io.confluent.kafka.serializers.AbstractKafkaAvroSerDe.<init>(AbstractKafkaAvroSerDe.java:46)
at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.<init>(AbstractKafkaAvroSerializer.java:37)
at io.confluent.kafka.serializers.KafkaAvroSerializer.<init>(KafkaAvroSerializer.java:32)
... 28 common frames omitted
Для consumer аналогичная, только с дисериализатором.
Зависимости в грэдле:
implementation ("org.springframework.cloud:spring-cloud-stream-binder-kafka:2.1.1.RELEASE")
implementation ("io.confluent:kafka-schema-registry-client:4.1.1")
implementation ("io.confluent:kafka-avro-serializer:4.1.1")
implementation ("org.apache.avro:avro-compiler:1.8.2")
С чем может быть связано?
Ответы (1 шт):
Автор решения: Sergei Kirjanov
→ Ссылка
Я читал, что для работы с авро-спецификациями у confluent используется реестр-сервис.
Сам использую protobuf.