Kafka - TimeoutException: Expiring 1 record(s) for art-0:120001 ms has passed since batch creation

文章目录

• 问题描述
• 原因分析
• Code

  

  ---

  问题描述

  报错如下：

  ....
  ....
  Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for art-0:120001 ms has passed since batch creation
  

  ---

  原因分析

  这种情况，肯定要先看网络问题嘛

  • 首先查看本机防火墙的配置

    结果都是关闭的 （建议开放特定端口）

    [root@localhost bin]# systemctl status firewalld.service
    ● firewalld.service - firewalld - dynamic firewall daemon
       Loaded: loaded (/usr/lib/systemd/system/firewalld.service; disabled; vendor preset: enabled)
       Active: inactive (dead)
         Docs: man:firewalld(1)
    [root@localhost bin]#
    

    不关闭防火墙，但是建议本机防火墙开放特定端口，可以使用如下命令 （使用root账户）

    firewall-cmd --zone=public --add-port=2181/tcp --permanent
    firewall-cmd --zone=public --add-port=9092/tcp --permanent
    firewall-cmd --reload
    firewall-cmd --list-ports
    

    比如

    [root@localhost bin]# systemctl status  firewalld
    ● firewalld.service - firewalld - dynamic firewall daemon
       Loaded: loaded (/usr/lib/systemd/system/firewalld.service; disabled; vendor preset: enabled)
       Active: active (running) since Thu 2023-10-26 10:13:10 CST; 2min 36s ago
         Docs: man:firewalld(1)
     Main PID: 40311 (firewalld)
        Tasks: 2
       Memory: 28.1M
       CGroup: /system.slice/firewalld.service
               └─40311 /usr/bin/python2 -Es /usr/sbin/firewalld --nofork --nopid
    Oct 26 10:13:10 localhost.localdomain systemd[1]: Starting firewalld - dynamic firewall daemon...
    Oct 26 10:13:10 localhost.localdomain systemd[1]: Started firewalld - dynamic firewall daemon.
    Oct 26 10:13:10 localhost.localdomain firewalld[40311]: WARNING: AllowZoneDrifting is enabled. This is considered an insecure configurati...it now.
    Hint: Some lines were ellipsized, use -l to show in full.
    [root@localhost bin]#
    [root@localhost bin]#
    [root@localhost bin]# firewall-cmd --zone=public --add-port=2181/tcp --permanent
    success
    [root@localhost bin]# firewall-cmd --zone=public --add-port=9092/tcp --permanent
    success
    [root@localhost bin]#
    [root@localhost bin]# firewall-cmd --list-ports
    [root@localhost bin]#
    [root@localhost bin]# firewall-cmd --reload
    success
    [root@localhost bin]# firewall-cmd --list-ports
    2181/tcp 9092/tcp
    [root@localhost bin]#
    

    ---

    • 接着看看kafka中间件的配置， 问题就在这里

      我并没有大改配置，具体的配置可参考 Kafak - 单机/集群快速安装指北(3.x版本)

      如下的配置并没有修改

      要解决这个问题，修改如上配置即可

      ---

      Code

      package com.artisan.pc;
      import org.apache.kafka.clients.producer.KafkaProducer;
      import org.apache.kafka.clients.producer.ProducerConfig;
      import org.apache.kafka.clients.producer.ProducerRecord;
      import org.apache.kafka.clients.producer.RecordMetadata;
      import org.apache.kafka.common.serialization.StringSerializer;
      import java.util.Properties;
      import java.util.concurrent.ExecutionException;
      /**
       * @author 小工匠
       * @version 1.0
       * @mark: show me the code , change the world
       */
      public class CustomProducer {
          public static void main(String[] args) throws ExecutionException, InterruptedException {
              // 1. 创建kafka生产者的配置对象
              Properties properties = new Properties();
              // 2. 给kafka配置对象添加配置信息
              properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.126.170:9092");
              // key,value序列化
              properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
              properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
              // 3. 创建kafka生产者对象
              KafkaProducer kafkaProducer = new KafkaProducer(properties);
              // 4. 调用send方法,发送消息
              for (int i = 0; i < 10; i++) {
              	// 同步阻塞 
                  RecordMetadata art = kafkaProducer.send(new ProducerRecord<>("art", "kafka-msg-" + i)).get();
                  System.out.println(art.topic());
                  System.out.println("over - " +i);
              }
              // 5. 关闭资源
              kafkaProducer.close();
          }
      }
          
      

      可以看消费者的控制台程序，输出正常