Kafka Spring Boot 消费者配置注意事项

秒速五厘米 2022-01-23 15:39 734阅读 0赞

## Kafka Spring Boot 消费者配置注意事项 ##

Kafka 使用Spring Boot集成时，配置Consumer时遇到运行时错误：  
java.io.EOFException: null  
或者  
Bootstrap broker … disconnected

如果配置了SASL参数，那么，恭喜你，这个问题，本篇博客可以解决你的问题。

由于SASL配置在Spring Boot集成时的配置是非标准的，所以如果配置文件不对，那么链接kafka服务器是链接就会断开，而不会提示是否是SASL参数问题，为了找到问题，调试好久才发现。

正确的配置方式是：(yaml格式)

spring:
      kafka:
        # 舆情系统Kafka参数配置
        consumer:
          #省略...
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          
          #这里是关键<----
          properties:
            security.protocol: SASL_PLAINTEXT
            sasl.mechanism: PLAIN
            sasl.kerberos.service.name: kafka
          #

上面配置中，properties就是问题的关键，必须使用，而且不能和前面一样使用“-”（减号）分隔，必须使用“.”（点）分隔这个配置，这个是非标准参数配置的特殊性。

## 参考资料 ##

#### Spring boot kafka 配置说明 ####

[官方文档链接][Link 1]

spring.kafka.consumer.key-deserializer= # Deserializer class for keys.
    spring.kafka.consumer.max-poll-records= # Maximum number of records returned in a single call to poll().
    spring.kafka.consumer.properties.*= # Additional consumer-specific properties used to configure the client.
    spring.kafka.consumer.ssl.key-password= # Password of the private key in the key store file.
    spring.kafka.consumer.ssl.key-store-location= # Location of the key store file.
    spring.kafka.consumer.ssl.key-store-password= # Store password for the key store file.

从上面可以看出properties的配置是没有在文档详细说明的。  
而代码中，需要的是"sasl.mechanism"配置项，所以，需要注意分隔符。

#### [Kafka源码参考未知][Kafka] ####

@ConfigurationProperties(prefix = "spring.kafka")
    public class KafkaProperties {
        public static class Consumer {
            /**
    		 * Additional consumer-specific properties used to configure the client.
    		 */
    		private final Map<String, String> properties = new HashMap<>();
        }
    }

通过上方代码可以看出，Spring boot只是把spring.kafka.consumer的properties参数赋值给消费者配置属性。  
如果通过存Java代码初始化，可以直接把Properties的sasl配置直接和其他配置对其，而不会有问题。  
具体的初始化逻辑可以通过断点调试来确认了。

[Link 1]: https://docs.spring.io/spring-boot/docs/current/reference/html/common-application-properties.html
[Kafka]: https://github.com/spring-projects/spring-boot/blob/b5d19b798eca7a6400a97e80d42601e027789aa6/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java#L217