diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java b/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java index f0c267a..e1675bc 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java @@ -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 producer, String partitionKeyStrategy) { this.appProperties = appProperties; diff --git a/haobang-security-dm/syslog-serve/src/main/resources/application.properties b/haobang-security-dm/syslog-serve/src/main/resources/application.properties index 18ff490..d1fce49 100644 --- a/haobang-security-dm/syslog-serve/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-serve/src/main/resources/application.properties @@ -86,6 +86,8 @@ spring.kafka.producer.properties.request.timeout.ms=30000 spring.kafka.producer.properties.delivery.timeout.ms=120000 spring.kafka.producer.properties.max.block.ms=10000 spring.kafka.producer.properties.buffer.memory=134217728 +# 允许投递加密后超过 Kafka 默认 1 MiB 的 Syslog;Broker/Topic 上限必须不小于此值。 +spring.kafka.producer.properties.max.request.size=${KAFKA_PRODUCER_MAX_REQUEST_SIZE:4194304} # event_id 使高流量设备的独立事件均匀分布到多个分区;device_id 仅用于兼容回滚。 syslog.kafka.partition-key-strategy=${SYSLOG_KAFKA_PARTITION_KEY_STRATEGY:event_id} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/kafka/KafkaProducerEventIdTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/kafka/KafkaProducerEventIdTest.java index 8c8f2d0..cab9855 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/kafka/KafkaProducerEventIdTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/kafka/KafkaProducerEventIdTest.java @@ -4,10 +4,12 @@ import com.haobang.config.AppProperties; import com.common.entity.DeviceDevice; import com.haobang.util.DeviceInfoUtil; import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.junit.jupiter.api.Test; import java.time.Instant; +import java.util.Properties; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -17,6 +19,21 @@ import static org.mockito.Mockito.mock; class KafkaProducerEventIdTest { + @Test + void maxRequestSizeIsPassedToTheManuallyConstructedKafkaProducer() { + Properties properties = new Properties(); + + kafkaProducer.applyMaxRequestSize(properties, 4 * 1024 * 1024); + + assertEquals(4 * 1024 * 1024, properties.get(ProducerConfig.MAX_REQUEST_SIZE_CONFIG)); + } + + @Test + void nonPositiveMaxRequestSizeFailsFast() { + assertThrows(IllegalArgumentException.class, + () -> kafkaProducer.applyMaxRequestSize(new Properties(), 0)); + } + @Test void eventIdDateUsesShanghaiTimeInsteadOfJvmDefaultTimeZone() { Instant shanghaiMidnight = Instant.parse("2026-08-10T16:00:00Z");