Kafka节点断开连接求助：生产者/消费者无法收发事件

出现以下日志报错，导致生产者无法发送事件、消费者无法接收事件：

2023-02-01 03:42:40.614  INFO 3524 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1] Node 1 disconnected.
2023-02-01 03:47:41.228  INFO 3524 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1] Node 0 disconnected.
2023-02-01 03:47:41.228  INFO 3524 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1] Cancelled in-flight METADATA request with correlation id 617 due to node 0 being disconnected (elapsed time since creation: 108ms, elapsed time since send: 108ms, request timeout: 2000ms)
2023-02-01 05:02:48.399  INFO 3524 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1] Node 2 disconnected.

生产者配置类 #

package com.bbh.ilab.aip.presentationservice.config

import com.bbh.ilab.aip.commons.kafka.BaseEvent
import com.bbh.ilab.aip.commons.kafka.EventPayload
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.common.config.SslConfigs
import org.apache.kafka.common.serialization.StringSerializer
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.kafka.core.ProducerFactory
import org.springframework.kafka.support.serializer.JsonSerializer


@Configuration
class KafkaProducerConfig(
    private val environmentProperties: EnvironmentProperties
) {

    @Bean
    fun producerFactory(): ProducerFactory<String, BaseEvent<EventPayload>> {
        val configProps = getStandardConfig()
        if (environmentProperties.kafka.sslEnabled) {
            addSslConfig(configProps)
        }
        return DefaultKafkaProducerFactory(configProps)
    }

    @Bean
    fun kafkaTemplate(): KafkaTemplate<String, BaseEvent<EventPayload>> {
        return KafkaTemplate(producerFactory())
    }

    private fun getStandardConfig(): MutableMap<String, Any> {
        val configProps: MutableMap<String, Any> = HashMap()
        configProps[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = environmentProperties.kafka.bootstrapUrl
        configProps[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
        configProps[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = JsonSerializer::class.java
        configProps[ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG] = environmentProperties.kafka.retry.deliveryTimeoutMs
        configProps[ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG] = environmentProperties.kafka.retry.requestTimeoutMs
        configProps[ProducerConfig.RETRY_BACKOFF_MS_CONFIG] = environmentProperties.kafka.retry.retryBackoffMs

        return configProps
    }

    private fun addSslConfig(configProps: MutableMap<String, Any>) {
        configProps["security.protocol"] = "SSL"
        configProps[SslConfigs.SSL_KEYSTORE_TYPE_CONFIG] = "PKCS12"
        configProps[SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG] = "PKCS12"
        configProps[SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG] = environmentProperties.kafka.keystorePath
        configProps[SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG] = environmentProperties.kafka.keystorePass
        configProps[SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG] = environmentProperties.kafka.truststorePath
        configProps[SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG] = environmentProperties.kafka.truststorePass
        configProps[SslConfigs.SSL_KEY_PASSWORD_CONFIG] = environmentProperties.kafka.keystorePass
    }
}

事件发送代码 #

fun sendSplitEvent(configDto: ConfigDto, domain: Domain, timestamp: LocalDateTime) {
    val payload = SplitColumnRequestPayload(
        configDto.projectId,
        domain,
        configDto.targetColumn,
        configDto.split
    )
    val event = SplitRequestEvent(UUID.randomUUID(), timestamp, payload)
    kafkaTemplate.send(environmentProperties.kafka.topic, event as BaseEvent<EventPayload>)
}

已针对相关配置进行调整，但问题仍未解决，盼技术支持。

内容的提问来源于stack exchange，提问作者Krzysztof Pasiek