生产环境丢消息后，如何复现Kafka TimeoutException异常？

生产环境出现消息丢失问题，需复现org.apache.kafka.common.errors.TimeoutException以验证修复方案（设置Ack=1并在send()后调用flush()）的有效性。

当前配置与代码 #

生产者工厂配置 #

public ProducerFactory<String, MyMessage> producerFacory() {
   Map<String, Object> configProps = kafkaPropperties.buildProducerProperties();
   configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, Integer.toString(512*1024)); // 试图触发超时
   return new DefaultKafkaProducerFactory<>(configProps);
}

发布者代码 #

public void myMethod() {
   // 创建消息的代码
   kafkaTemplate.send(myTopic, someId, message);
   Thread.sleep(60000); // 尝试强制触发超时
}

已尝试操作 #

• 移除Linger配置
• 将linger设置为10ms
• 将linger设置为60000ms
• 调小BATCH_SIZE_CONFIG
• 调大BATCH_SIZE_CONFIG

测试场景：应用在20秒至2分钟间随机触发消息，持续运行2小时，单条消息大小1.4Kb，预期触发超时但未复现。

复现TimeoutException的有效方案 #

要复现该异常，核心是让生产者无法在delivery.timeout.ms（默认30秒）内完成消息发送或收到预期ACK，可从以下方向操作：

1. 模拟网络异常 #

在生产者所在机器用工具添加高延迟或丢包规则，让消息传输时间远超超时阈值：

# Linux环境下用tc命令添加25秒延迟+50%丢包
tc qdisc add dev eth0 root netem delay 25000ms loss 50%

2. 调整生产者超时相关配置 #

主动缩短超时时间并延长linger时间，让凑批等待超过超时限制：

configProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "10000"); // 设为10秒超时
configProps.put(ProducerConfig.LINGER_MS_CONFIG, "15000"); // linger时间超过超时时间

3. 让Kafka集群无法正常响应 #

• 停止目标Broker：找到消息对应分区的主Broker并停止服务，生产者无法连接目标节点，超时后触发异常。
• 设置不合理的ACK规则：将生产者acks设为all，同时把目标主题的min.insync.replicas设为大于当前可用同步副本数。例如主题有3个副本但仅1个Broker在线：

  configProps.put(ProducerConfig.ACKS_CONFIG, "all");
  

  同时修改主题配置：

  kafka-configs.sh --bootstrap-server localhost:9092 --alter --topic myTopic --add-config min.insync.replicas=2
  

4. 调整消息发送逻辑 #

去掉Thread.sleep，改为每秒发送1条1.4Kb消息，同时设置较大的linger和batch size：

configProps.put(ProducerConfig.LINGER_MS_CONFIG, "30000"); // 30秒凑批
configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, "10485760"); // 10MB批大小

生产者会一直等待凑批，直到delivery.timeout.ms到期触发超时。

内容的提问来源于stack exchange，提问作者Ben Fourie