Kafka是一个分布式的、可分区的、可复制的消息系统，下面是Kafka的几个基本术语：

  1. Kafka将消息以topic为单位进行归纳；
  2. 将向Kafka topic发布消息的程序成为producers；
  3. 将预订topics并消费消息的程序成为consumer；
  4. Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker。

producers通过网络将消息发送到Kafka集群，集群向消费者提供消息，如下图所示：

Spring Boot整合Kafka - 图1

创建一个topic时，可以指定partitions（分区）数目，partitions数越多，其吞吐量也越大，但是需要的资源也越多，同时也会导致更高的不可用性，kafka在接收到producers发送的消息之后，会根据均衡策略将消息存储到不同的partitions中：

Spring Boot整合Kafka - 图2

在每个partitions中，消息以顺序存储，最晚接收的的消息会最后被消费。

producers在向kafka集群发送消息的时候，可以通过指定partitions来发送到指定的partitions中。也可以通过指定均衡策略来将消息发送到不同的partitions中。如果不指定，就会采用默认的随机均衡策略，将消息随机的存储到不同的partitions中。

在consumer消费消息时，kafka使用offset来记录当前消费的位置：

Spring Boot整合Kafka - 图3

在kafka的设计中，可以有多个不同的group来同时消费同一个topic下的消息，如图，我们有两个不同的group同时消费，他们的的消费的记录位置offset各不项目，不互相干扰。

对于一个group而言，consumer的数量不应该多于partitions的数量，因为在一个group中，每个partitions至多只能绑定到一个consumer上，即一个consumer可以消费多个partitions，一个partitions只能给一个consumer消费。因此，若一个group中的consumer数量大于partitions数量的话，多余的consumer将不会收到任何消息。

Spring Boot整合Kafka - 图4

Kafka安装使用

这里演示在Windows下Kafka安装与使用。Kafka下载地址：http://kafka.apache.org/downloads，选择二进制文件下载（Binary downloads），然后解压即可。

Kafka的配置文件位于config目录下，因为Kafka集成了Zookeeper（Kafka存储消息的地方），所以config目录下除了有Kafka的配置文件server.properties外，还可以看到一个Zookeeper配置文件zookeeper.properties：

Spring Boot整合Kafka - 图5

打开server.properties，将broker.id的值修改为1，每个broker的id都必须设置为Integer类型，且不能重复。Zookeeper的配置保持默认即可。

接下来开始使用Kafka。

启动Zookeeper

在Windows下执行下面这些命令可能会出现找不到或无法加载主类的问题，解决方案可参考：

  1. CLASSPATH配置有误，应该是：.;%JAVA_HOME%\lib\dt.jar;%JAVA_HOME%\lib\tools.jar; 最前面是“小数点”不是“逗号”。
  2. java环境由JRE换成JDK的话，需要修改kafka_2.12-1.0.0\bin\windows\kafka-run-class.bat文件。

具体修改内容是，将：
set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp %CLASSPATH% %KAFKA_OPTS% %
改为：set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp “%CLASSPATH%” %KAFKA_OPTS% %
%CLASSPATH%要用双引号。

在Kafka根目录下使用cmd执行下面这条命令，启动ZK：

  1. bin\windows\zookeeper-server-start.bat config\zookeeper.properties

在Linux下，可以使用后台进程的方式启动ZK：

  1. bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

安装Kafka 2.11 参考：使用Docker启动KafKa

启动Kafka

执行下面这条命令启动Kafka：

  1. bin\windows\kafka-server-start.bat config\server.properties

Linux对应命令：

  1. bin/kafka-server-start.sh config/server.properties

当看到命令行打印如下信息，说明启动完毕:

Spring Boot整合Kafka - 图6

注意：新版Kafka已经全部将**--zookeeper localhost:2181**换成 **--bootstrap-server broker主机名称（或IP):端口（一般为9092）**

创建Topic

执行下面这条命令创建一个Topic

  1. bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

这条命令的意思是，创建一个Topic到ZK（指定ZK的地址），副本个数为1，分区数为1，Topic的名称为test。

Linux对应的命令为:

  1. bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

创建好后我们可以查看Kafka里的Topic列表：

  1. bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

Spring Boot整合Kafka - 图7

可看到目前只包含一个我们刚创建的test Topic。

Linux对应的命令为：

  1. bin/kafka-topics.sh --list --zookeeper localhost:2181

查看test Topic的具体信息：

  1. bin\windows\kafka-topics.bat --describe --zookeeper localhost:2181 --topic test

Spring Boot整合Kafka - 图8

Linux对应的命令为：

  1. bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

生产消息和消费消息

启动Producers

  1. bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test

9092为生产者的默认端口号。这里启动了生产者，准备往test Topic里发送数据。

Linux下对应的命令为：

  1. bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

启动Consumers

接着启动一个消费者用于消费生产者生产的数据，新建一个cmd窗口，输入下面这条命令：

  1. bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning

from-beginning表示从头开始读取数据。

Linux下对应的命令为：

  1. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

启动好生产者和消费者后我们在生产者里生产几条数据:

Spring Boot整合Kafka - 图9

消费者成功接收到数据：

Spring Boot整合Kafka - 图10

Spring Boot整合Kafaka

上面简单介绍了Kafka的使用，下面我们开始在Spring Boot里使用Kafka。

新建一个Spring Boot项目，版本为2.1.3.RELEASE，并引入如下依赖:

  1.  <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-web</artifactId>
  4. </dependency>
  5. <dependency>
  6.     <groupId>org.springframework.kafka</groupId>
  7.     <artifactId>spring-kafka</artifactId>
  8. </dependency>

生产者配置

新建一个Java配置类KafkaProducerConfig，用于配置生产者：

  1. @Configuration
  2. public class KafkaProducerConfig {
  4.     @Value("${spring.kafka.bootstrap-servers}")
  5.     private String bootstrapServers;
  7.     @Bean
  8.     public ProducerFactory<String, String> producerFactory() {
  9.         Map<String, Object> configProps = new HashMap<>();
  10.         configProps.put(
  11.                 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  12.                 bootstrapServers);
  13.         configProps.put(
  14.                 ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
  15.                 StringSerializer.class);
  16.         configProps.put(
  17.                 ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
  18.                 StringSerializer.class);
  19.         return new DefaultKafkaProducerFactory<>(configProps);
  20.     }
  22.     @Bean
  23.     public KafkaTemplate<String, String> kafkaTemplate() {
  24.         return new KafkaTemplate<>(producerFactory());
  25.     }
  26. }

首先我们配置了一个producerFactory，方法里配置了Kafka Producer实例的策略。bootstrapServers为Kafka生产者的地址，我们在配置文件application.yml里配置它：

  1. spring:
  2.   kafka:
  3.     bootstrap-servers: localhost:9092

ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG和ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG指定了key，value序列化策略，这里指定为Kafka提供的StringSerializer，因为我们暂时只发送简单的String类型的消息。

接着我们使用producerFactory配置了kafkaTemplate，其包含了发送消息的便捷方法，后面我们就用这个对象来发送消息。

发布消息

配置好生产者，我们就可以开始发布消息了。

新建一个SendMessageController：

  1. @RestController
  2. public class SendMessageController {
  4.     @Autowired
  5.     private KafkaTemplate<String, String> kafkaTemplate;
  7.     @GetMapping("send/{message}")
  8.     public void send(@PathVariable String message) {
  9.         this.kafkaTemplate.send("test", message);
  10.     }
  11. }

我们注入了kafkaTemplate对象，key-value都为String类型，并通过它的send方法来发送消息。其中test为Topic的名称，上面我们已经使用命令创建过这个Topic了。

send方法是一个异步方法，我们可以通过回调的方式来确定消息是否发送成功，我们改造SendMessageController：

  1. @RestController
  2. public class SendMessageController {
  4.     private Logger logger = LoggerFactory.getLogger(this.getClass());
  6.     @Autowired
  7.     private KafkaTemplate<String, String> kafkaTemplate;
  9.     @GetMapping("send/{message}")
  10.     public void send(@PathVariable String message) {
  11.         ListenableFuture<SendResult<String, String>> future = this.kafkaTemplate.send("test", message);
  12.         future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
  13.             @Override
  14.             public void onSuccess(SendResult<String, String> result) {
  15.                 logger.info("成功发送消息：{}，offset=[{}]", message, result.getRecordMetadata().offset());
  16.             }
  18.             @Override
  19.             public void onFailure(Throwable ex) {
  20.                 logger.error("消息：{} 发送失败，原因：{}", message, ex.getMessage());
  21.             }
  22.         });
  23.     }
  24. }

消息发送成功后，会回调onSuccess方法，发送失败后回调onFailure方法。

消费者配置

接着我们来配置消费者，新建一个Java配置类KafkaConsumerConfig：

  1. @EnableKafka
  2. @Configuration
  3. public class KafkaConsumerConfig {
  5.     @Value("${spring.kafka.bootstrap-servers}")
  6.     private String bootstrapServers;
  8.     @Value("${spring.kafka.consumer.group-id}")
  9.     private String consumerGroupId;
  11.     @Value("${spring.kafka.consumer.auto-offset-reset}")
  12.     private String autoOffsetReset;
  14.     @Value("${spring.kafka.consumer.session.timeout.ms}")
  15.     private String sessionTimeoutMs;
  17.     @Bean
  18.     public ConsumerFactory<String, String> consumerFactory() {
  19.         Map<String, Object> props = new HashMap<>();
  20.         props.put(
  21.                 ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  22.                 bootstrapServers);
  23.         props.put(
  24.                 ConsumerConfig.GROUP_ID_CONFIG,
  25.                 consumerGroupId);
  26.         props.put(
  27.                 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
  28.                 autoOffsetReset);
  29.         props.put(
  30.                 ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,
  31.                 sessionTimeoutMs);
  32.         props.put(
  33.                 ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
  34.                 StringDeserializer.class);
  35.         props.put(
  36.                 ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
  37.                 StringDeserializer.class);
  38.         return new DefaultKafkaConsumerFactory<>(props);
  39.     }
  41.     @Bean
  42.     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
  43.         ConcurrentKafkaListenerContainerFactory<String, String> factory
  44.                 = new ConcurrentKafkaListenerContainerFactory<>();
  45.         factory.setConsumerFactory(consumerFactory());
  46.         return factory;
  47.     }
  48. }

consumerGroupId和autoOffsetReset需要在application.yml里配置：

  1. spring:
  2.   kafka:
  3.     consumer:
  4.       group-id: test-consumer
  5.       auto-offset-reset: latest
  6.       session:
  7.        timeout:
  8.          ms: 6000

其中group-id将消费者进行分组（你也可以不进行分组），组名为test-consumer，并指定了消息读取策略，包含四个可选值：

Spring Boot整合Kafka - 图11

  • earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
  • latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
  • none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
  • exception:直接抛出异常

在KafkaConsumerConfig中我们配置了ConsumerFactory和KafkaListenerContainerFactory。当这两个Bean成功注册到Spring IOC容器中后，我们便可以使用@KafkaListener注解来监听消息了。

配置类上需要@EnableKafka注释才能在Spring托管Bean上检测@KafkaListener注解。

消息消费

配置好消费者，我们就可以开始消费消息了，新建KafkaMessageListener：

  1. @Component
  2. public class KafkaMessageListener {
  4.     private Logger logger = LoggerFactory.getLogger(this.getClass());
  6.     @KafkaListener(topics = "test", groupId = "test-consumer")
  7.     public void listen(String message) {
  8.         logger.info("接收消息: {}", message);
  9.     }
  10. }

我们通过@KafkaListener注解来监听名称为test的Topic，消费者分组的组名为test-consumer。

演示

启动Spring Boot项目，启动过程中，控制台会输出Kafka的配置，启动好后，访问http://localhost:8080/send/hello,mrbird，控制台输出如下：

Spring Boot整合Kafka - 图12

@KafkaListener详解

@KafkaListener除了可以指定Topic名称和分组id外，我们还可以同时监听来自多个Topic的消息:

  1. @KafkaListener(topics = "topic1, topic2")

我们还可以通过@Header注解来获取当前消息来自哪个分区（partitions）:

  1. @KafkaListener(topics = "test", groupId = "test-consumer")
  2. public void listen(@Payload String message,
  3.                    @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
  4.     logger.info("接收消息: {}，partition：{}", message, partition);
  5. }

重启项目，再次访问http://localhost:8080/send/hello,mrbird，控制台输出如下：

Spring Boot整合Kafka - 图13

因为我们没有进行分区，所以test Topic只有一个区，下标为0。

我们可以通过@KafkaListener来指定只接收来自特定分区的消息：

  1. @KafkaListener(groupId = "test-consumer",
  2.         topicPartitions = @TopicPartition(topic = "test",
  3.                 partitionOffsets = {
  4.                         @PartitionOffset(partition = "0", initialOffset = "0")
  5.             }))
  6. public void listen(@Payload String message,
  7.                    @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
  8.     logger.info("接收消息: {}，partition：{}", message, partition);
  9. }

如果不需要指定initialOffset，上面代码可以简化为：

  1. @KafkaListener(groupId = "test-consumer", 
  2.     topicPartitions = @TopicPartition(topic = "test", partitions = { "0", "1" }))

消息过滤器

我们可以为消息监听添加过滤器来过滤一些特定的信息。我们在消费者配置类KafkaConsumerConfig的kafkaListenerContainerFactory方法里配置过滤规则：

  1. @Bean
  2. public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
  3.     ConcurrentKafkaListenerContainerFactory<String, String> factory
  4.             = new ConcurrentKafkaListenerContainerFactory<>();
  5.     factory.setConsumerFactory(consumerFactory());
  6.     // ------- 过滤配置 --------
  7.     factory.setRecordFilterStrategy(
  8.             r -> r.value().contains("fuck")
  9.     );
  10.     return factory;
  11. }

setRecordFilterStrategy接收RecordFilterStrategy<K, V>，他是一个函数式接口：

  1. public interface RecordFilterStrategy<K, V> {
  2.     boolean filter(ConsumerRecord<K, V> var1);
  3. }

所以我们用lambda表达式指定了上面这条规则，即如果消息内容包含fuck这个粗鄙之语的时候，则不接受消息。

配置好后我们重启项目，分别发送下面这两条请求：

  1. http://localhost:8080/send/fuck,mrbird
  2. http://localhost:8080/send/love,mrbird

观察控制台：

Spring Boot整合Kafka - 图14

可以看到，fuck,mrbird这条消息没有被接收。

发送复杂的消息

截至目前位置我们只发送了简单的字符串类型的消息，我们可以自定义消息转换器来发送复杂的消息。

定义消息实体

创建一个Message类：

  1. public class Message implements Serializable {
  2.     private static final long serialVersionUID = 6678420965611108427L;
  4.     private String from;
  6.     private String message;
  8.     public Message() {
  10.     }
  12.     public Message(String from, String message) {
  13.         this.from = from;
  14.         this.message = message;
  15.     }
  17.     @Override
  18.     public String toString() {
  19.         return "Message{" +
  20.                 "from='" + from + '\'' +
  21.                 ", message='" + message + '\'' +
  22.                 '}';
  23.     }
  25.     // get set 略
  26. }

改造消息生产者配置

  1. @Configuration
  2. public class KafkaProducerConfig {
  4.     @Value("${spring.kafka.bootstrap-servers}")
  5.     private String bootstrapServers;
  7.     @Bean
  8.     public ProducerFactory<String, Message> producerFactory() {
  9.         Map<String, Object> configProps = new HashMap<>();
  10.         configProps.put(
  11.                 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  12.                 bootstrapServers);
  13.         configProps.put(
  14.                 ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
  15.                 StringSerializer.class);
  16.         configProps.put(
  17.                 ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
  18.                 JsonSerializer.class);
  19.         return new DefaultKafkaProducerFactory<>(configProps);
  20.     }
  22.     @Bean
  23.     public KafkaTemplate<String, Message> kafkaTemplate() {
  24.         return new KafkaTemplate<>(producerFactory());
  25.     }
  26. }

我们将value序列化策略指定为了Kafka提供的JsonSerializer，并且kafkaTemplate返回类型为KafkaTemplate<String, Message>。

发送新的消息

在SendMessageController里发送复杂的消息:

  1. @Autowired
  2. private KafkaTemplate<String, Message> kafkaTemplate;
  4. @GetMapping("send/{message}")
  5. public void sendMessage(@PathVariable String message) {
  6.     this.kafkaTemplate.send("test", new Message("mrbird", message));
  7. }

修改消费者配置

修改消费者配置KafkaConsumerConfig：

  1. @EnableKafka
  2. @Configuration
  3. public class KafkaConsumerConfig {
  5.     @Value("${spring.kafka.bootstrap-servers}")
  6.     private String bootstrapServers;
  8.     @Value("${spring.kafka.consumer.group-id}")
  9.     private String consumerGroupId;
  11.     @Value("${spring.kafka.consumer.auto-offset-reset}")
  12.     private String autoOffsetReset;
  14.     @Bean
  15.     public ConsumerFactory<String, Message> consumerFactory() {
  16.         Map<String, Object> props = new HashMap<>();
  17.         props.put(
  18.                 ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
  19.                 bootstrapServers);
  20.         props.put(
  21.                 ConsumerConfig.GROUP_ID_CONFIG,
  22.                 consumerGroupId);
  23.         props.put(
  24.                 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
  25.                 autoOffsetReset);
  26.         return new DefaultKafkaConsumerFactory<>(
  27.                 props,
  28.                 new StringDeserializer(),
  29.                 new JsonDeserializer<>(Message.class));
  30.     }
  32.     @Bean
  33.     public ConcurrentKafkaListenerContainerFactory<String, Message> kafkaListenerContainerFactory() {
  34.         ConcurrentKafkaListenerContainerFactory<String, Message> factory
  35.                 = new ConcurrentKafkaListenerContainerFactory<>();
  36.         factory.setConsumerFactory(consumerFactory());
  37.         return factory;
  38.     }
  39. }

修改消息监听

修改KafkaMessageListener：

  1. @KafkaListener(topics = "test", groupId = "test-consumer")
  2. public void listen(Message message) {
  3.     logger.info("接收消息: {}", message);
  4. }

重启项目，访问http://localhost:8080/send/hello，控制台输出如下：

Spring Boot整合Kafka - 图15

更多配置

  1. spring.kafka.admin.client-id= # ID to pass to the server when making requests. Used for server-side logging.
  2. spring.kafka.admin.fail-fast=false # Whether to fail fast if the broker is not available on startup.
  3. spring.kafka.admin.properties.*= # Additional admin-specific properties used to configure the client.
  4. spring.kafka.admin.ssl.key-password= # Password of the private key in the key store file.
  5. spring.kafka.admin.ssl.key-store-location= # Location of the key store file.
  6. spring.kafka.admin.ssl.key-store-password= # Store password for the key store file.
  7. spring.kafka.admin.ssl.key-store-type= # Type of the key store.
  8. spring.kafka.admin.ssl.protocol= # SSL protocol to use.
  9. spring.kafka.admin.ssl.trust-store-location= # Location of the trust store file.
  10. spring.kafka.admin.ssl.trust-store-password= # Store password for the trust store file.
  11. spring.kafka.admin.ssl.trust-store-type= # Type of the trust store.
  12. spring.kafka.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connections to the Kafka cluster. Applies to all components unless overridden.
  13. spring.kafka.client-id= # ID to pass to the server when making requests. Used for server-side logging.
  14. spring.kafka.consumer.auto-commit-interval= # Frequency with which the consumer offsets are auto-committed to Kafka if 'enable.auto.commit' is set to true.
  15. spring.kafka.consumer.auto-offset-reset= # What to do when there is no initial offset in Kafka or if the current offset no longer exists on the server.
  16. spring.kafka.consumer.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connections to the Kafka cluster. Overrides the global property, for consumers.
  17. spring.kafka.consumer.client-id= # ID to pass to the server when making requests. Used for server-side logging.
  18. spring.kafka.consumer.enable-auto-commit= # Whether the consumer's offset is periodically committed in the background.
  19. spring.kafka.consumer.fetch-max-wait= # Maximum amount of time the server blocks before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by "fetch-min-size".
  20. spring.kafka.consumer.fetch-min-size= # Minimum amount of data the server should return for a fetch request.
  21. spring.kafka.consumer.group-id= # Unique string that identifies the consumer group to which this consumer belongs.
  22. spring.kafka.consumer.heartbeat-interval= # Expected time between heartbeats to the consumer coordinator.
  23. spring.kafka.consumer.key-deserializer= # Deserializer class for keys.
  24. spring.kafka.consumer.max-poll-records= # Maximum number of records returned in a single call to poll().
  25. spring.kafka.consumer.properties.*= # Additional consumer-specific properties used to configure the client.
  26. spring.kafka.consumer.ssl.key-password= # Password of the private key in the key store file.
  27. spring.kafka.consumer.ssl.key-store-location= # Location of the key store file.
  28. spring.kafka.consumer.ssl.key-store-password= # Store password for the key store file.
  29. spring.kafka.consumer.ssl.key-store-type= # Type of the key store.
  30. spring.kafka.consumer.ssl.protocol= # SSL protocol to use.
  31. spring.kafka.consumer.ssl.trust-store-location= # Location of the trust store file.
  32. spring.kafka.consumer.ssl.trust-store-password= # Store password for the trust store file.
  33. spring.kafka.consumer.ssl.trust-store-type= # Type of the trust store.
  34. spring.kafka.consumer.value-deserializer= # Deserializer class for values.
  35. spring.kafka.jaas.control-flag=required # Control flag for login configuration.
  36. spring.kafka.jaas.enabled=false # Whether to enable JAAS configuration.
  37. spring.kafka.jaas.login-module=com.sun.security.auth.module.Krb5LoginModule # Login module.
  38. spring.kafka.jaas.options= # Additional JAAS options.
  39. spring.kafka.listener.ack-count= # Number of records between offset commits when ackMode is "COUNT" or "COUNT_TIME".
  40. spring.kafka.listener.ack-mode= # Listener AckMode. See the spring-kafka documentation.
  41. spring.kafka.listener.ack-time= # Time between offset commits when ackMode is "TIME" or "COUNT_TIME".
  42. spring.kafka.listener.client-id= # Prefix for the listener's consumer client.id property.
  43. spring.kafka.listener.concurrency= # Number of threads to run in the listener containers.
  44. spring.kafka.listener.idle-event-interval= # Time between publishing idle consumer events (no data received).
  45. spring.kafka.listener.log-container-config= # Whether to log the container configuration during initialization (INFO level).
  46. spring.kafka.listener.monitor-interval= # Time between checks for non-responsive consumers. If a duration suffix is not specified, seconds will be used.
  47. spring.kafka.listener.no-poll-threshold= # Multiplier applied to "pollTimeout" to determine if a consumer is non-responsive.
  48. spring.kafka.listener.poll-timeout= # Timeout to use when polling the consumer.
  49. spring.kafka.listener.type=single # Listener type.
  50. spring.kafka.producer.acks= # Number of acknowledgments the producer requires the leader to have received before considering a request complete.
  51. spring.kafka.producer.batch-size= # Default batch size.
  52. spring.kafka.producer.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connections to the Kafka cluster. Overrides the global property, for producers.
  53. spring.kafka.producer.buffer-memory= # Total memory size the producer can use to buffer records waiting to be sent to the server.
  54. spring.kafka.producer.client-id= # ID to pass to the server when making requests. Used for server-side logging.
  55. spring.kafka.producer.compression-type= # Compression type for all data generated by the producer.
  56. spring.kafka.producer.key-serializer= # Serializer class for keys.
  57. spring.kafka.producer.properties.*= # Additional producer-specific properties used to configure the client.
  58. spring.kafka.producer.retries= # When greater than zero, enables retrying of failed sends.
  59. spring.kafka.producer.ssl.key-password= # Password of the private key in the key store file.
  60. spring.kafka.producer.ssl.key-store-location= # Location of the key store file.
  61. spring.kafka.producer.ssl.key-store-password= # Store password for the key store file.
  62. spring.kafka.producer.ssl.key-store-type= # Type of the key store.
  63. spring.kafka.producer.ssl.protocol= # SSL protocol to use.
  64. spring.kafka.producer.ssl.trust-store-location= # Location of the trust store file.
  65. spring.kafka.producer.ssl.trust-store-password= # Store password for the trust store file.
  66. spring.kafka.producer.ssl.trust-store-type= # Type of the trust store.
  67. spring.kafka.producer.transaction-id-prefix= # When non empty, enables transaction support for producer.
  68. spring.kafka.producer.value-serializer= # Serializer class for values.
  69. spring.kafka.properties.*= # Additional properties, common to producers and consumers, used to configure the client.
  70. spring.kafka.ssl.key-password= # Password of the private key in the key store file.
  71. spring.kafka.ssl.key-store-location= # Location of the key store file.
  72. spring.kafka.ssl.key-store-password= # Store password for the key store file.
  73. spring.kafka.ssl.key-store-type= # Type of the key store.
  74. spring.kafka.ssl.protocol= # SSL protocol to use.
  75. spring.kafka.ssl.trust-store-location= # Location of the trust store file.
  76. spring.kafka.ssl.trust-store-password= # Store password for the trust store file.
  77. spring.kafka.ssl.trust-store-type= # Type of the trust store.
  78. spring.kafka.streams.application-id= # Kafka streams application.id property; default spring.application.name.
  79. spring.kafka.streams.auto-startup=true # Whether or not to auto-start the streams factory bean.
  80. spring.kafka.streams.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connections to the Kafka cluster. Overrides the global property, for streams.
  81. spring.kafka.streams.cache-max-size-buffering= # Maximum memory size to be used for buffering across all threads.
  82. spring.kafka.streams.client-id= # ID to pass to the server when making requests. Used for server-side logging.
  83. spring.kafka.streams.properties.*= # Additional Kafka properties used to configure the streams.
  84. spring.kafka.streams.replication-factor= # The replication factor for change log topics and repartition topics created by the stream processing application.
  85. spring.kafka.streams.ssl.key-password= # Password of the private key in the key store file.
  86. spring.kafka.streams.ssl.key-store-location= # Location of the key store file.
  87. spring.kafka.streams.ssl.key-store-password= # Store password for the key store file.
  88. spring.kafka.streams.ssl.key-store-type= # Type of the key store.
  89. spring.kafka.streams.ssl.protocol= # SSL protocol to use.
  90. spring.kafka.streams.ssl.trust-store-location= # Location of the trust store file.
  91. spring.kafka.streams.ssl.trust-store-password= # Store password for the trust store file.
  92. spring.kafka.streams.ssl.trust-store-type= # Type of the trust store.
  93. spring.kafka.streams.state-dir= # Directory location for the state store.
  94. spring.kafka.template.default-topic= # Default topic to which messages are sent.