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

---

问题描述

报错如下：

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

---

原因分析

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

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

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

[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]#

1
2
3
4
5
6
7
8

不关闭防火墙，但是建议本机防火墙开放特定端口，可以使用如下命令 （使用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
1
2
3
4

比如

[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]#

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

---

• 接着看看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<String, String> kafkaProducer = new KafkaProducer<String, String>(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();

    }

}
    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

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