Input Kafka avro - registry authentication

Elastic Stack Logstash
1

Olá,

I have some issues related with registry authentication.

My conf:

input {
  kafka {
    id => "kafka_xxx"
    bootstrap_servers => "xxxxx:9094"
    topics => ["XXX_XX_XXX_XX"]

    auto_offset_reset => "earliest"
    security_protocol => "SASL_SSL"
    sasl_mechanism => "SCRAM-SHA-512"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='xxxxx' password='xxxxx';"
    group_id => "XXX"
    client_id => "XXX"
    ssl_truststore_location => "config/CA_CHAIN.truststore.jks"
    ssl_truststore_password => "passss"

    schema_registry_url => "https://registry_domain:8081"
    schema_registry_key => "USER_INFO"
    schema_registry_secret => "application:pass"

    schema_registry_ssl_keystore_password => "passss"
    schema_registry_ssl_keystore_location => "config/CA_CHAIN.truststore.jks"
    schema_registry_ssl_keystore_type => "jks"

    schema_registry_ssl_truststore_type => "jks"
    schema_registry_ssl_truststore_location => "config/CA_CHAIN.truststore.jks"
    schema_registry_ssl_truststore_password => "passss"
    ##schema_registry_validation => "skip"

    #key_deserializer_class => "org.apache.kafka.common.serialization.StringDeserializer"
    #value_deserializer_class => "io.confluent.kafka.serializers.KafkaAvroDeserializer"

    codec => json {}
  }
}

Error:

[2025-02-14T13:08:16,640][ERROR][logstash.inputs.kafka ][xxx_kafka][kafka_xxx] Unable to poll Kafka consumer {:kafka_error_message=>#<Java::OrgApacheKafkaCommonErrors::RecordDeserializationException: Error deserializing VALUE for partition TOPIC_XXXX-0 at offset 368. If needed, please seek past the record to continue consumption.>, :cause=>#<Java::OrgApacheKafkaCommonErrors::AuthenticationException: Error retrieving Avro value schema for id 471>}

If I remove the registry and force value_deserializer_class => "io.confluent.kafka.serializers.KafkaAvroDeserializer" I receive a message :

Missing required configuration "schema.registry.url" which has no default value.

I need to use KafkaAvroDeserializer.