Error retrieving Avro schema Schema not found error code 40403

Question

I have the below config of kafka

I tried to get the subject by version and i get like below

curl -u reader:reader -s "https://schema-registry1:8081/subjects/add_bank_account/versions/1/" | jq
{
  "subject": "add_bank_account",
  "version": 1,
  "id": 46,
  "schema": "{\"type\":\"record\",\"name\":\"CreateBankAccount\",\"namespace\":\"com.test.serviceprovider.avro\",\"fields\":[{\"name\":\"exerciceFrameworkId\",\"type\":\"string\"},{\"name\":\"agreementFileId\",\"type\":\"string\"},{\"name\":\"conventionCode\",\"type\":\"string\"}]}"
}

I generate the CreateBankAccount using the plugin avro-maven-plugin

CreateBankAccount createBankAccount = CreateBankAccount.newBuilder()
    .setExerciceFrameworkId(event.exerciceFrameworkId)
    .setAgreementFileId(event.agreementFileId)
    .setConventionCode(convention.getConventionCode()).build();

Then i pushed a message to the topic with a string key and an avro serialized payload, but i have an error

MessageChannel messageChannel = kafkaStreams.addBankAccount();
messageChannel.send(MessageBuilder.withPayload(createBankAccount)
   .setHeader(KafkaHeaders.MESSAGE_KEY, UUID.randomUUID().toString()).build());

Tracetrack:

Caused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro schema: {"type":"record","name":"CreateBankAccount","namespace":"com.test.serviceprovider.avro","fields":[{"name":"exerciceFrameworkId","type":{"type":"string","avro.java.string":"String"}},{"name":"agreementFileId","type":{"type":"string","avro.java.string":"String"}},{"name":"conventionCode","type":{"type":"string","avro.java.string":"String"}}]} Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema not found; error code: 40403 at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:292) at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:352) at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:437) at io.confluent.kafka.schemaregistry.client.rest.RestService.lookUpSubjectVersion(RestService.java:424) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getIdFromRegistry(CachedSchemaRegistryClient.java:243) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getId(CachedSchemaRegistryClient.java:389) at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:89) at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:60) at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:62) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:894) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:856) at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:592) at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:404) at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:241) at org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler.handleRequestMessage(KafkaProducerMessageHandler.java:410) at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:127) at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:177) at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder$SendingHandler.handleMessageInternal(AbstractMessageChannelBinder.java:1035) at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:177) at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115) at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133) at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106) at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73) at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453) at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:403) at com.test.entities.handlers.ConventionEntityManager.on(ConventionEntityManager.java:168) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.axonframework.messaging.annotation.AnnotatedMessageHandlingMember.handle(AnnotatedMessageHandlingMember.java:144) at org.axonframework.messaging.annotation.AnnotatedHandlerInspector$NoMoreInterceptors.handle(AnnotatedHandlerInspector.java:372) at org.axonframework.eventhandling.AnnotationEventHandlerAdapter.handle(AnnotationEventHandlerAdapter.java:94) at org.axonframework.eventhandling.SimpleEventHandlerInvoker.handle(SimpleEventHandlerInvoker.java:112) at org.axonframework.eventhandling.MultiEventHandlerInvoker.handle(MultiEventHandlerInvoker.java:89) at org.axonframework.eventhandling.AbstractEventProcessor.lambda$null$1(AbstractEventProcessor.java:165) at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:57) at org.axonframework.messaging.interceptors.CorrelationDataInterceptor.handle(CorrelationDataInterceptor.java:65) at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:55) at org.axonframework.eventhandling.TrackingEventProcessor.lambda$new$1(TrackingEventProcessor.java:185) at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:55) at org.axonframework.eventhandling.AbstractEventProcessor.lambda$processInUnitOfWork$2(AbstractEventProcessor.java:173) at org.axonframework.messaging.unitofwork.BatchingUnitOfWork.executeWithResult(BatchingUnitOfWork.java:86) at org.axonframework.eventhandling.AbstractEventProcessor.processInUnitOfWork(AbstractEventProcessor.java:159) at org.axonframework.eventhandling.TrackingEventProcessor.processBatch(TrackingEventProcessor.java:462) at org.axonframework.eventhandling.TrackingEventProcessor.processingLoop(TrackingEventProcessor.java:294) at org.axonframework.eventhandling.TrackingEventProcessor$TrackingSegmentWorker.run(TrackingEventProcessor.java:1016) at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.run(TrackingEventProcessor.java:1162) at java.lang.Thread.run(Thread.java:748)

ProducerConfig values:

KafkaAvroSerializerConfig values:

Answer 1

I have a similar error. The problem is solved when I add this properties use.latest.version=true into my config of producer-properties

Answer 2

You're using TopicNameStrategy, so the subject name in the Registry must be <topicName>-value

You could also just set auto.register.schemas=true rather than register the schema on your own with some custom name.