Почему я получаю ошибку (SSLError) certificate verify failed при отправке сообщения в AVRO в Kafka?

У меня есть Kafka консьюмер, который может получать сообщения только в кодировке AVRO. Теперь я пытаюсь реализовать продюсера, который отправляет сообщения в кодировке AVRO в топик. Сообщения - это логи, генерируемые системой в файл. Понял, что могу читать логи и отправлять их с помощью Logstash и Filebeat.

Настроил сбор и изменение логов, но теперь не могу решить эту проблему. У меня есть такой output.

output {
  kafka {
    topic_id => "topic"
    bootstrap_servers => "servers"
    client_id => "id"
    security_protocol => "SASL_SSL"
    sasl_mechanism => "SCRAM-SHA-512"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='username' password='password';"
    ssl_truststore_location => "/etc/kafka.client.truststore.jks"
    ssl_truststore_password => "password"
    value_serializer => "org.apache.kafka.common.serialization.ByteArraySerializer"
    codec => json
  }
}

Он нормально работает и я могу читать json сообщения из топика. Но это было только для теста - json использовать не получится. Я сделал output ниже с помощью avro_schema_registry, но теперь logstash падает с ошибкой.

[logstash.javapipeline][main] Pipeline worker error, the pipeline will be stopped {:pipeline_id=>"main", :error=>"(SSLError) certificate verify failed", :exception=>Java::OrgJrubyExceptions::StandardError...

Как её можно исправить?

output {
  kafka {
    topic_id => "topic"
    bootstrap_servers => "servers"
    client_id => "id"
    security_protocol => "SASL_SSL"
    sasl_mechanism => "SCRAM-SHA-512"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='username_AAA' password='password_AAA';"
    ssl_truststore_location => "/etc/kafka.client.truststore.jks"
    ssl_truststore_password => "password"
    value_serializer => "org.apache.kafka.common.serialization.ByteArraySerializer"
    codec => avro_schema_registry {
      endpoint => "endpoint"
      schema_id => 1111
      register_schema => false
      username => "username_AAA"
      password => "password_AAA"
    }
  }
}

Может я неправильно понимаю как работает AVRO или SSL? Для консьюмера нужно, чтобы сообщение имело закодированный идентификатор схемы, потому что он работает с реестром схем. Поэтому в моем продюсере его нужно передавать. Но почему возвращается ошибка, если я уже добавил сертификат выше и он работал? И почему мне нужно обязательно передавать эндпоинт реестра схем? Может есть какой-то обходной путь, с которым я могу без avro_schema_registry передавать закодированные id схемы, саму схему и сообщение?