org.apache.kafka.clients. Producer.KafkaProducer#doSend 抛出 OOM 错误 | Kafka模板

Question

我正在使用 KafkaTemplate 向某个主题生成消息，但在 KafkaProducer 调用 doSend() 时出现 OOM 错误。不确定是否有关系。 Pod 被分配了 2Gi 的内存，我们正在批量处理 200 条消息，处理大约 10K 条消息。

下面的代码是全部出错的地方。

final ProducerRecord<String, byte[]> record =
    new ProducerRecord(message.getTopicIdentifier(), message.getKey(), message.getData());

message.getMetaData().getKeyValueMap().forEach((key, value) -> record.headers().add(key, value.getBytes()));
ListenableFuture<SendResult<String, byte[]>> listenableFuture =  kafkaTemplate.send(record);
listenableFuture.addCallback(new ListenableFutureCallback<>() {

    @Override
    public void onSuccess(SendResult<String, byte[]> result) {
        log.debug("Message : {} published to Kafka successful", message.getId());
    }
    @Override
    public void onFailure(Throwable ex) {
        log.error("[onFailure] Error processing message: {}", message.getId(), ex);
    }
});

MessageDTO 和其他相关类就像

public class Metadata {

    private Map<String,String> keyValueMap;
}

public class MessageDTO {

    private Long id;
    private byte[] data;
    private String key;
    private String status;
    private String reason;
    private String topicIdentifier;
    private Date createdDate;
    private Date updateDate;
    private Metadata metaData;
}

下面是生产者配置和允许 pod 使用的资源

spring:
    kafka:
      producer:
        bootstrapServers: **server**
        acks: all
        compressionType: lz4
        retries: 5

  resources:
    limits:
      cpu: 1000m
      memory: 2048Mi
    requests:
      cpu: 4m
      memory: 128Mi

Answer 1

您正在使用 200 个记录批次。您可以尝试将批处理大小减少到 10 甚至 1，并检查它是否停止抛出 OOM - 然后逐渐增加批处理大小，同时观察内存指标，为一个应用程序实例找到正确的批处理大小。然后您可以纵向扩展或横向扩展。

还要检查虚拟机是否配置正确并使用可用内存——这可能允许每个应用程序实例具有更大的批处理大小。