Kafka 大消息处理

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
// 单请求上限：必须 >= 最大消息 + 批次开销
props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 20 * 1024 * 1024);
// 单批上限：Kafka 会取 max(batch.size, 单条消息大小)
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 20 * 1024 * 1024);
// 缓冲总量：至少能装下几个大消息，否则 send 阻塞
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 256 * 1024 * 1024);

# broker 接收上限
message.max.bytes=20971520
# 副本同步上限：必须 >= message.max.bytes
replica.fetch.max.bytes=20971520
# 每个分区返回给消费者的上限（broker 侧软限制）
fetch.max.bytes=52428800

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
// 单分区单次返回上限：必须 >= 最大消息
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 20 * 1024 * 1024);
// 单次请求总上限
props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 52_428_800);
// 单次 poll 的记录数（软上限，不切割批次）
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);

max.partition.fetch.bytes >= message.max.bytes
replica.fetch.max.bytes   >= message.max.bytes
max.request.size          >= 最大消息 + 开销
fetch.max.bytes           >= max.partition.fetch.bytes

# 文本类大消息：zstd 兼顾压缩比与速度
compression.type=zstd
# 二进制大消息：关掉压缩
# compression.type=none

# 大消息场景：不攒批，发出去就好
linger.ms=0

生产者                    对象存储               Kafka
  │                         │                    │
  ├── 上传大文件 ──────────► │                    │
  │   (S3/MinIO)            │                    │
  │                         │                    │
  ├── 发送指针消息 ─────────────────────────────► │
  │   {"bucket":"...", "key":"...", "size":...}  │
  │                                              │
消费者 ◄──────────────────────────────────────────┘
  │  收到指针
  ├── 按指针从对象存储拉取 ◄───┘
  └── 处理内容

public class ClaimCheckProducer {
    private final KafkaProducer<String, String> producer;
    private final S3Client s3;
    private final ObjectMapper mapper = new ObjectMapper();

    public void send(String topic, String key, byte[] payload) throws Exception {
        String objectKey = "kafka-payloads/" + UUID.randomUUID() + ".bin";

        // 1. 先上传大内容，失败则整个操作失败，不会产生悬空指针
        s3.putObject(b -> b.bucket("app-blobs").key(objectKey)
                            .metadata(Map.of("sha256", sha256(payload))),
                     RequestBody.fromBytes(payload));

        // 2. 再发指针消息
        ClaimCheck ref = new ClaimCheck("app-blobs", objectKey,
                                        payload.length, sha256(payload));
        producer.send(new ProducerRecord<>(topic, key, mapper.writeValueAsString(ref)),
                      (meta, ex) -> {
                          if (ex != null) {
                              // 发送失败要清理已上传的对象，避免孤儿文件
                              s3.deleteObject(b -> b.bucket("app-blobs").key(objectKey));
                          }
                      });
    }
}

public void consume(ConsumerRecord<String, String> record) throws Exception {
    ClaimCheck ref = mapper.readValue(record.value(), ClaimCheck.class);

    try (ResponseInputStream<GetObjectResponse> in = s3.getObject(
            b -> b.bucket(ref.bucket()).key(ref.key()))) {
        byte[] payload = in.readAllBytes();

        // 校验完整性：对象存储可能被误删或覆盖
        if (!sha256(payload).equals(ref.sha256())) {
            throw new IntegrityException("内容哈希不匹配: " + ref.key());
        }
        process(payload);
    } catch (NoSuchKeyException e) {
        // 内容已被生命周期规则删除 —— 需要死信或人工介入
        deadLetter(record, e);
    }
}

原始大消息 8 MB
   ├── chunk 0: {"msgId":"m1","seq":0,"total":4,"data":"..."}  2 MB
   ├── chunk 1: {"msgId":"m1","seq":1,"total":4,"data":"..."}  2 MB
   ├── chunk 2: {"msgId":"m1","seq":2,"total":4,"data":"..."}  2 MB
   └── chunk 3: {"msgId":"m1","seq":3,"total":4,"data":"..."}  2 MB

String msgId = UUID.randomUUID().toString();
int chunkSize = 1 * 1024 * 1024;   // 每片 1MB，安全落在默认上限内

for (int seq = 0; seq * chunkSize < payload.length; seq++) {
    int from = seq * chunkSize;
    int to = Math.min(from + chunkSize, payload.length);
    byte[] slice = Arrays.copyOfRange(payload, from, to);

    String envelope = mapper.writeValueAsString(new Chunk(msgId, seq,
        (payload.length + chunkSize - 1) / chunkSize, slice));

    // 同一 key → 同一分区 → 顺序保证
    producer.send(new ProducerRecord<>("large-payloads", msgId, envelope));
}

public class ChunkAssembler {
    // msgId -> 已收到的分片；必须有上限，否则内存会被慢消费拖垮
    private final Map<String, TreeMap<Integer, byte[]>> pending =
        new LinkedHashMap<>(16, 0.75f, true) {
            @Override
            protected boolean removeEldestEntry(Map.Entry<String, TreeMap<Integer, byte[]>> e) {
                return size() > MAX_PENDING;   // 超过 1000 个未完成消息就淘汰最旧的
            }
        };

    public Optional<byte[]> accept(Chunk chunk) {
        TreeMap<Integer, byte[]> parts =
            pending.computeIfAbsent(chunk.msgId(), k -> new TreeMap<>());
        parts.put(chunk.seq(), chunk.data());

        if (parts.size() < chunk.total()) {
            return Optional.empty();   // 还没收齐
        }
        pending.remove(chunk.msgId());

        ByteArrayOutputStream out = new ByteArrayOutputStream();
        parts.values().forEach(b -> out.writeBytes(b));
        return Optional.of(out.toByteArray());
    }
}

// 定时清理超时未收齐的消息
scheduler.scheduleAtFixedRate(() -> {
    long cutoff = System.currentTimeMillis() - 60_000;
    pending.entrySet().removeIf(e -> e.getValue().isEmpty() /* 或按时间戳 */);
}, 30, 30, TimeUnit.SECONDS);

# 查 topic 的实际消息上限（可能被 topic 级配置覆盖）
kafka-configs.sh --bootstrap-server kafka:9092 \
  --entity-type topics --entity-name orders --describe

# 查最大消息尺寸的估算（用 kafka-run-class 的 dump-log-segment）
kafka-dump-log.sh --files /var/lib/kafka/data/orders-0/00000000000000000000.log \
  --print-data-log | awk '{print length($0)}' | sort -n | tail -1