Kubernetes环境下Kafka Streams应用OOM问题排查求助

问题概述 #

在Kubernetes环境中运行的Kafka Streams应用出现Out-Of-Memory（OOM）问题，Pod内存占用接近5Gi的限制，但Java堆内存平均仅使用1.01Gi，无法定位内存占用来源，需排查原因并解决。

环境与配置信息 #

• Kafka Streams版本：3.1.1
• Pod内存限制：5Gi
• Java最大堆内存（Xmx）：4Gi
• Pod实际内存使用：4.75GiB
• 平均堆内存使用：1.01GiB
• 平均非堆内存使用：221MiB
• 平均老年代GC大小：1GiB
• 临时存储使用、请求与限制：[附图]

应用配置代码 #

Properties properties = new Properties();
properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
properties.put(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, "DEBUG");
properties.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, KafkaStreamsExceptionHandler.class);
properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);

properties.put(StreamsConfig.APPLICATION_ID_CONFIG, streamsName);
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
properties.put("transactional.id.expiration.ms", 1500000000);

业务拓扑代码 #

KTable<Windowed<String>, PayLoadDTO> multiplePayment =
        sourceStream.mapValues(PayLoadDTO::new)
                .groupBy((key, value) -> value.getLoginFailureStreamDTO().getPayload().getLoginId(), Grouped.with(Serdes.String(),
                        PaymentSerdes.PayLoadDTOSerde()))
                .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(super.getWindowTime()), Duration.ZERO))
                .reduce(PayLoadDTO::aggregate,
                        Materialized.<String, PayLoadDTO, WindowStore<Bytes, byte[]>>as("device")
                                .withKeySerde(Serdes.String()).withValueSerde(PaymentSerdes.PayLoadDTOSerde()));

multiplePayment.toStream()
        .mapValues((key, value) -> createSinkDTO(value))
        .map((key, value) -> KeyValue.pair(String.valueOf(value.getPayload().getUid()), value))
        .to(super.getTopologyProperties().getSinkTopicName(), Produced.with(Serdes.String(), PaymentSerdes.PaymentSinkSerde()));

内存过高原因分析 #

1. 堆外内存占用超标：Pod总内存4.75Gi，而堆+非堆仅占用约1.23Gi，剩余3.5Gi左右为堆外内存。Kafka Streams依赖的RocksDB状态存储、Kafka客户端直接内存、JNI调用、线程栈等均会占用堆外内存，这是本次OOM的核心原因。
   • 窗口状态存储默认使用RocksDB，其块缓存、写缓冲区均为堆外内存，若窗口数据量较大，这部分内存会持续增长。
   • Kafka Producer/Consumer的网络IO缓冲区、SSL相关内存也属于堆外范畴，未计入Java堆统计。
2. 窗口配置不合理：窗口的grace period设为Duration.ZERO，虽会立即清理过期窗口，但如果窗口时长（super.getWindowTime()）过大，窗口存续期间会累积大量数据，导致RocksDB堆外内存占用飙升。
3. 事务ID过期时间设置过大：transactional.id.expiration.ms设为17天（1500000000ms），远大于默认7天，会导致Kafka保留过多事务元数据，额外占用内存。
4. JVM堆外内存未限制：未设置MaxDirectMemorySize等参数，Direct Memory可能无限增长；线程栈默认1M/线程，若Kafka Streams线程数较多，这部分堆外内存也会累积。

解决方案 #

1. 定位堆外内存具体来源 #

• 执行jcmd <pid> VM.native_memory summary命令，查看堆外内存各部分占用情况，明确是RocksDB、Direct Memory还是线程栈等导致的内存超标。

2. 优化RocksDB状态存储配置 #

• 通过自定义RocksDB配置调整堆外内存占用：

  properties.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, CustomRocksDBConfig.class);
  

  实现CustomRocksDBConfig类，调整块缓存、写缓冲区大小：

  public class CustomRocksDBConfig implements RocksDBConfigSetter {
      @Override
      public void setConfig(String storeName, Options options, Map<String, Object> configs) {
          // 减小块缓存大小，例如设为256MB
          options.setBlockCache(new LRUCache(256 * 1024 * 1024));
          // 减小写缓冲区大小，例如设为32MB
          options.setWriteBufferSize(32 * 1024 * 1024);
          // 减少列族的写缓冲区数量
          options.setMaxWriteBufferNumber(2);
      }
  }
  

3. 调整窗口配置 #

• 若业务允许，设置合理的grace period（如5分钟），避免窗口数据在存续期过度累积；若窗口时长过大，考虑缩短窗口时间，降低单窗口数据量。

  .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(super.getWindowTime()), Duration.ofMinutes(5)))
  

4. 修正事务ID过期时间 #

• 将transactional.id.expiration.ms调整为默认值7天（604800000ms）或业务合理值：

  properties.put("transactional.id.expiration.ms", 604800000);
  

5. 限制JVM堆外内存 #

• 添加JVM参数限制Direct Memory大小，例如：

  -XX:MaxDirectMemorySize=1G
  

• 若线程数较多，减小线程栈大小：

  -Xss256k
  

6. 调整Pod与JVM内存分配 #

• 当前JVM堆占4Gi，留给堆外仅1Gi，远不足实际需求。建议将Xmx调整为3Gi，给堆外预留2Gi空间，或根据堆外实际占用情况调整Pod内存限制。

7. 检查临时存储 #

• 结合附图查看临时存储使用情况，若RocksDB磁盘存储占用过高，可能导致内存映射文件（mmap）占用额外堆外内存。需确保临时存储有足够空间，或定期清理过期的状态存储文件。

内容的提问来源于stack exchange，提问作者채희원