Приложение Spring не подключается к Kafka с SSL - PullRequest
0 голосов
/ 30 мая 2019

У меня есть загрузочное приложение Spring с очень простым производителем кафки.Все отлично работает, если я подключаюсь к кластеру kafka без шифрования.Но время ожидания, если я пытаюсь подключиться к кластеру kafka с SSL.Есть ли какая-то другая конфигурация, которая мне нужна в производителе, или какое-то другое свойство, которое мне нужно определить, чтобы позволить Spring правильно использовать все конфигурации?

У меня установлены следующие свойства

    spring.kafka.ssl.key-store-type=jks
    spring.kafka.ssl.trust-store-location=file:/home/ec2-user/truststore.jks
    spring.kafka.ssl.trust-store-password=test1234
    spring.kafka.ssl.key-store-location=file:/home/ec2-user/keystore.jks
    spring.kafka.ssl.key-store-password=test1234
    logging.level.org.apache.kafka=debug
    server.ssl.key-password=test1234
    spring.kafka.ssl.key-password=test1234
    spring.kafka.producer.client-id=sym
    spring.kafka.admin.ssl.protocol=ssl

При печати следующего результата в качестве ProducerConfig при запуске приложения

    o.a.k.clients.producer.ProducerConfig    : ProducerConfig values:
    acks = 1
    batch.size = 16384
    bootstrap.servers = [broker1.kafka.allypoc.com:9093, broker3.kafka.allypoc.com:9093, broker4.kafka.allypoc.com:9093, broker5.kafka.allypoc.com:9093]
    buffer.memory = 33554432
    client.dns.lookup = default
    client.id = sym
    compression.type = none
    connections.max.idle.ms = 540000
    delivery.timeout.ms = 120000
    enable.idempotence = false
    interceptor.classes = []
    key.serializer = class org.apache.kafka.common.serialization.StringSerializer
    linger.ms = 0
    max.block.ms = 60000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 32768
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 2147483647
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = https
    ssl.key.password = [hidden]
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = /home/ec2-user/keystore.jks
    ssl.keystore.password = [hidden]
    ssl.keystore.type = jks
    ssl.protocol = ssl
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = /home/ec2-user/truststore.jks
    ssl.truststore.password = [hidden]
    ssl.truststore.type = JKS
    transaction.timeout.ms = 60000
    transactional.id = null
    value.serializer = class org.apache.kafka.common.serialization.StringSerializer

Мой производитель очень прост:

    @Service
    public class Producer {

        private final KafkaTemplate<String, String> kafkaTemplate;

        public Producer(KafkaTemplate<String, String> kafkaTemplate) {
            this.kafkaTemplate = kafkaTemplate;
        }

        void sendMessage(String topic, String message) {
            this.kafkaTemplate.send(topic, message);
        }

        void sendMessage(String topic, String key, String message) {
            this.kafkaTemplate.send(topic, key, message);
        }
    }

При подключении к kafka с помощью SSL появляется исключение TimeoutException, говорящее Topic symbols not present in metadata after 60000 ms. Если я включаю журналы отладки, я получаю это неоднократно, зацикливая всех своих брокеров.

2019-05-29 20:10:25.768 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.clients.NetworkClient   : [Producer clientId=sym] Completed connection to node -4. Fetching API versions.
    2019-05-29 20:10:25.768 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.clients.NetworkClient   : [Producer clientId=sym] Initiating API versions fetch from node -4.
    2019-05-29 20:10:25.768 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.clients.NetworkClient   : [Producer clientId=sym] Initialize connection to node 10.25.77.13:9093 (id: -3 rack: null) for sending metadata request
    2019-05-29 20:10:25.768 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.clients.NetworkClient   : [Producer clientId=sym] Initiating connection to node 10.25.77.13:9093 (id: -3 rack: null) using address /10.25.77.13
    2019-05-29 20:10:25.994 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.common.metrics.Metrics  : Added sensor with name node--3.bytes-sent
    2019-05-29 20:10:25.996 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.common.metrics.Metrics  : Added sensor with name node--3.bytes-received
    2019-05-29 20:10:25.997 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.common.metrics.Metrics  : Added sensor with name node--3.latency
    2019-05-29 20:10:25.998 DEBUG 1381 --- [rk-thread | sym] o.apache.kafka.common.network.Selector   : [Producer clientId=sym] Created socket with SO_RCVBUF = 32768, SO_SNDBUF = 131072, SO_TIMEOUT = 0 to node -3
    2019-05-29 20:10:26.107 DEBUG 1381 --- [rk-thread | sym] o.apache.kafka.common.network.Selector   : [Producer clientId=sym] Connection with /10.25.75.151 disconnected

    java.io.EOFException: null
        at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:119) ~[kafka-clients-2.1.1.jar!/:na]
        at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:381) ~[kafka-clients-2.1.1.jar!/:na]
        at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:342) ~[kafka-clients-2.1.1.jar!/:na]
        at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:609) ~[kafka-clients-2.1.1.jar!/:na]
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:541) ~[kafka-clients-2.1.1.jar!/:na]
        at org.apache.kafka.common.network.Selector.poll(Selector.java:467) ~[kafka-clients-2.1.1.jar!/:na]
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:535) ~[kafka-clients-2.1.1.jar!/:na]
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:311) ~[kafka-clients-2.1.1.jar!/:na]
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:235) ~[kafka-clients-2.1.1.jar!/:na]
        at java.base/java.lang.Thread.run(Thread.java:835) ~[na:na]

    2019-05-29 20:10:26.108 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.clients.NetworkClient   : [Producer clientId=sym] Node -1 disconnected.
    2019-05-29 20:10:26.110 DEBUG 1381 --- [rk-thread | sym] org.apache.kafka.clients.NetworkClient   : [Producer clientId=sym] Completed connection to node -3. Fetching API versions.

1 Ответ

1 голос
/ 30 мая 2019

В настройках производителя security.protocol должен быть установлен на SSL. Вы также можете попробовать установить ssl.endpoint.identification.algirithm = "", чтобы отключить проверку имени хоста сертификата в случае, если это проблема. Кроме этого, было бы полезно посмотреть конфигурацию брокера Kafka.

...