How to connect logstash to Kafka over SSL?

Question

linux

Hello.
I have two kafka machines clustered together with SSL protection configured.
There is a machine with logstash where I want to connect to read topics from kafka.

Kafkas work with each other via SSL, but logstash cannot be connected as a consumer.

Connection error

[2021-08-26T17:30:39,847][ERROR][logstash.inputs.kafka    ][main][08d5bfe8d4808c6648f30d48f5b01d884e19b90b752cbde2fe9ff5f01e048d8d] Unable to poll Kafka consumer {:kafka_error_message=>org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed, :cause=>javax.net.ssl.SSLHandshakeException: java.security.cert.CertificateException: No subject alternative names present}

That is, an alternate name is required.
I tried to generate keys in two ways
But as a result, the SSL handshake does not work.

I found in one answer to add ssl.endpoint.identification.algorithm= so that alternative names are not checked, but this did not help. I only have IP addresses of machines.

spoiler

openssl req  -newkey rsa:2048 -nodes -keyout kafka-1.key -x509 -days 365 -out kafka-1.pem
openssl req -subj "/CN=${HOS1}/" -newkey rsa:2048 -nodes -keyout kafka-1.key -x509 -days 365 -out kafka-1.pem

Machine settings:

kafka-1

broker.id=0
listeners=PLAINTEXT://${HOST1}:9092, SSL://${HOST1}:9093
advertised.listeners=PLAINTEXT://${HOST1}:9092,SSL://${HOST1}:9093
zookeeper.connect=${HOST1}:2181
log.dirs=/tmp/kafka-logs

server.1=${HOST1}:2888:3888
server.2=${HOST2}:2888:3888

ssl.endpoint.identification.algorithm=

security.protocol=SSL
ssl.keystore.location=/opt/kafka/security/server.keystore.p12
ssl.keystore.password=Passwd123

ssl.truststore.location=/opt/kafka/security/client.truststore.jks
ssl.truststore.password=Passwd123

kafka-2

broker.id=1
listeners=PLAINTEXT://${HOST2}:9092, SSL://${HOST2}:9093
advertised.listeners=PLAINTEXT://${HOST2}:9092,SSL://${HOST2}:9093
zookeeper.connect=${HOST2}:2181
log.dirs=/tmp/kafka-logs


server.1=${HOST1}:2888:3888
server.2=${HOST2}:2888:3888

ssl.endpoint.identification.algorithm=

security.protocol=SSL
ssl.keystore.location=/opt/kafka/security/server.keystore.p12
ssl.keystore.password=Passwd123

ssl.truststore.location=/opt/kafka/security/client.truststore.jks
ssl.truststore.password=Passwd123

logstash /etc/logstash/conf.d/apache.conf

input {
  kafka {
    bootstrap_servers => "${HOST1}:9093, ${HOST2}:9093"
    topics => "apache"
    consumer_threads => 2
    group_id => "logstash-0"
    security_protocol => "SSL"
    ssl_keystore_location => "logstash-1.keystore.p12"
    ssl_keystore_password => "Passwd123"
    ssl_key_password => "Passwd123"
    ssl_truststore_location => "/etc/logstash/kafka-cluster.truststore.jks"
    ssl_truststore_password => "Passwd123"
    }
}
filter {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    date {
      match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
    }
  geoip {
      source => "clientip"
    }
}
output {
  stdout { codec => rubydebug }
}

I have generated keys, certificates and stores.
Added pem certificates of all servers to all servers in TrustStore.

trustStore on logstash

/etc/logstash # keytool -keystore kafka-cluster.truststore.jks -list

Enter keystore password:  
Keystore type: PKCS12
Keystore provider: SUN

Your keystore contains 3 entries

kafka-1, 26.08.2021, trustedCertEntry, 
Certificate fingerprint (SHA-256): F5:5F:CD:CA:AF:22:92:FB:0C:F3:55:D3:82:AE:EE:1D:70:4E:52:C7:6F:61:AE:38:24:83:BC:59:8B:22:B1:6D
kafka-2, 26.08.2021, trustedCertEntry, 
Certificate fingerprint (SHA-256): 88:49:6F:D6:9A:35:E4:31:59:AB:94:9F:65:E0:E8:BA:50:FB:7F:7F:6D:3F:53:63:88:C0:9C:90:58:D1:29:80
logstash-1, 26.08.2021, trustedCertEntry, 
Certificate fingerprint (SHA-256): 4B:F1:93:5C:09:A4:A3:14:71:2A:58:B2:F1:B1:71:89:BC:F7:A6:82:03:F6:17:73:8A:91:94:2C:E5:EF:04:D4

Answer 1

in the losgstash config connect to the PLAINTEXT port

bootstrap_servers => "${HOST1}:9092, ${HOST2}:9092"