[fix] kafka消息最大调整为4M
This commit is contained in:
@@ -53,6 +53,7 @@ public class kafkaProducer {
|
||||
@Value("${spring.kafka.producer.compression-type:lz4}") String compressionType,
|
||||
@Value("${spring.kafka.producer.properties.max.block.ms:10000}") int maxBlockMs,
|
||||
@Value("${spring.kafka.producer.properties.buffer.memory:134217728}") long bufferMemory,
|
||||
@Value("${spring.kafka.producer.properties.max.request.size:4194304}") int maxRequestSize,
|
||||
@Value("${syslog.kafka.partition-key-strategy:event_id}") String partitionKeyStrategy) {
|
||||
this.appProperties = appProperties;
|
||||
this.partitionKeyStrategy = PartitionKeyStrategy.parse(partitionKeyStrategy);
|
||||
@@ -73,12 +74,20 @@ public class kafkaProducer {
|
||||
// Dispatcher 已限制全局在途数;Producer 缓冲耗尽时快速返回失败,让事件留在 WAL 重试。
|
||||
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
|
||||
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
|
||||
applyMaxRequestSize(props, maxRequestSize);
|
||||
// [SYSLOG-ALARM][KPROD][已修复] linger、batch 与压缩恢复跨消息批量能力,吞吐优化不改变 Broker ACK 边界。
|
||||
// [SYSLOG-ALARM][KPROD][已修复] Producer 只在应用启动时构造一次,单条热路径不再创建网络线程和连接。
|
||||
this.producer = new KafkaProducer<>(props);
|
||||
log.info("Kafka Producer 已启用分区策略: strategy={}", this.partitionKeyStrategy.configValue);
|
||||
}
|
||||
|
||||
static void applyMaxRequestSize(Properties props, int maxRequestSize) {
|
||||
if (maxRequestSize <= 0) {
|
||||
throw new IllegalArgumentException("spring.kafka.producer.properties.max.request.size 必须大于 0");
|
||||
}
|
||||
props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize);
|
||||
}
|
||||
|
||||
/** 测试入口:复用真实的分区键选择和 ProducerRecord 构造逻辑,不建立 Kafka 网络连接。 */
|
||||
kafkaProducer(AppProperties appProperties, Producer<String, String> producer, String partitionKeyStrategy) {
|
||||
this.appProperties = appProperties;
|
||||
|
||||
Reference in New Issue
Block a user