OpenShift部署KCache运行12小时出现CacheInitializationException崩溃求解

API 探路者

2026-10-2

问题描述

我有一个运行KCache的应用，该应用包含多个线程，每个线程分别从一个topic拉取消息，然后将记录保存到各自的kcache中。每个线程会定期读取一组已存储的记录进行分析。该应用以容器形式部署在OpenShift上，初始阶段运行状态正常。
但在运行约12小时后，应用崩溃并抛出如下错误信息：

Exception in thread "main" io.kcache.exceptions.CacheInitializationException: Failed trying to create or validate my-topic-kcache
    at io.kcache.KafkaCache.createOrVerifyTopic(KafkaCache.java:350)
    at io.kcache.KafkaCache.init(KafkaCache.java:267)
    at utils.Cache.<init>(Cache.kt:40)
    at consumers.MyThread.<init>(MyThread.kt:39)
    at ApplicationKt.module(Application.kt:88)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.base/java.lang.reflect.Method.invoke(Unknown Source)
    at kotlin.reflect.jvm.internal.calls.CallerImpl$Method.callMethod(CallerImpl.kt:97)
    at kotlin.reflect.jvm.internal.calls.CallerImpl$Method$Static.call(CallerImpl.kt:106)
    at kotlin.reflect.jvm.internal.KCallableImpl.call(KCallableImpl.kt:108)
    at kotlin.reflect.jvm.internal.KCallableImpl.callDefaultMethod$kotlin_reflection(KCallableImpl.kt:159)
    at kotlin.reflect.jvm.internal.KCallableImpl.callBy(KCallableImpl.kt:112)
    at io.ktor.server.engine.internal.CallableUtilsKt.callFunctionWithInjection(CallableUtils.kt:119)
    at io.ktor.server.engine.internal.CallableUtilsKt.executeModuleFunction(CallableUtils.kt:36)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading$launchModuleByName$1.invoke(ApplicationEngineEnvironmentReloading.kt:332)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading$launchModuleByName$1.invoke(ApplicationEngineEnvironmentReloading.kt:331)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading.avoidingDoubleStartupFor(ApplicationEngineEnvironmentReloading.kt:356)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading.launchModuleByName(ApplicationEngineEnvironmentReloading.kt:331)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading.access$launchModuleByName(ApplicationEngineEnvironmentReloading.kt:30)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading$instantiateAndConfigureApplication$1.invoke(ApplicationEngineEnvironmentReloading.kt:312)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading$instantiateAndConfigureApplication$1.invoke(ApplicationEngineEnvironmentReloading.kt:310)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading.avoidingDoubleStartup(ApplicationEngineEnvironmentReloading.kt:338)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading.instantiateAndConfigureApplication(ApplicationEngineEnvironmentReloading.kt:310)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading.createApplication(ApplicationEngineEnvironmentReloading.kt:143)
    at io.ktor.server.engine.ApplicationEngineEnvironmentReloading.start(ApplicationEngineEnvironmentReloading.kt:277)
    at io.ktor.server.netty.NettyApplicationEngine.start(NettyApplicationEngine.kt:174)
    at io.ktor.server.netty.EngineMain.main(EngineMain.kt:26)
    at ApplicationKt.main(Application.kt:33)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1632808693387, tries=1, nextAllowedTryMs=-9223372036854775709) timed out at 9223372036854775807 after 1 attempt(s)
    at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
    at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
    at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:104)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:272)
    at io.kcache.KafkaCache.createOrVerifyTopic(KafkaCache.java:336)
    ... 29 more
Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1632808693387, tries=1, nextAllowedTryMs=-9223372036854775709) timed out at 9223372036854775807 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited. Call: listTopics
{"@timestamp":"2021-09-28T07:57:57.759+02:00","@version":"1","message":"Thread kafka-cache-reader-thread-thread1-kcache exiting with uncaught exception: ","logger_name":"io.kcache.utils.ShutdownableThread","thread_name":"kafka-cache-reader-thread-thread1-kcache","level":"ERROR","level_value":40000,"stack_trace":"java.lang.OutOfMemoryError: Java heap space\n"}
Exception in thread "thread1" java.lang.OutOfMemoryError: Java heap space
Exception in thread "kafka-cache-reader-thread-thread1-kcache" java.lang.OutOfMemoryError: Java heap space

缓存配置如下：

val cacheProps = Properties()
            cacheProps[KafkaCacheConfig.KAFKACACHE_BOOTSTRAP_SERVERS_CONFIG] =
                appConfig.propertyOrNull("kafka.bootstrapServers")?.getString()
            cacheProps[KafkaCacheConfig.KAFKACACHE_GROUP_ID_CONFIG] =
                appConfig.propertyOrNull("kafka.consumerGroupIdPrefix")?.getString() + "-$name-kcache"
            cacheProps[KafkaCacheConfig.KAFKACACHE_TOPIC_CONFIG] =
                appConfig.propertyOrNull("applicationId")?.getString() + "$name-kcache"
            cacheProps[KafkaCacheConfig.KAFKACACHE_SECURITY_PROTOCOL_CONFIG] =
                appConfig.propertyOrNull("kafka.securityProtocol")?.getString()

请问是否有人遇到过类似问题，或者有对应的解决方案？

解决方案

首先，错误日志中的*Java堆内存溢出（OOM）*是根因，后续的Kafka AdminClient超时、缓存初始化失败都是OOM引发的次生错误：堆内存耗尽后，Kafka客户端线程、缓存读取线程无法正常执行甚至被终止，导致listTopics、缓存初始化等操作超时失败。
可按以下步骤优化：

内存配置优化
KCache默认会将对应topic的全量数据加载到JVM堆内存中，你的场景下多线程各自维护独立KCache实例，相当于多份全量数据同时占用堆内存，运行12小时数据累积到阈值后触发OOM。首先调整OpenShift容器的JVM启动参数，添加-Xms<初始堆大小> -Xmx<最大堆大小>，建议最大堆大小至少设置为所有KCache预估总数据量的1.5倍以上，同时给容器预留至少2G的非堆内存空间，避免容器被系统OOM Killer直接杀掉。
如果业务不需要保留全量历史数据，可以给KCache对应的Kafka topic设置消息过期时间、单topic最大容量，也可以在KCache配置中添加淘汰策略：cacheProps[KafkaCacheConfig.KAFKACACHE_MAX_CACHE_SIZE_CONFIG] = <最大缓存条目数>，超过阈值后旧数据会被自动淘汰，避免内存无限增长。
KCache实例复用优化
如果多个线程处理的topic数据有重叠，优先复用同一个KCache实例，不要每个线程单独创建，避免重复加载相同数据占用多倍内存。如果必须拆分独立实例，建议将不同KCache实例分散部署到不同Pod，不要在单个Pod内启动过多实例。
Kafka客户端参数调优
当前配置未设置Kafka AdminClient的超时时间，建议添加配置cacheProps[KafkaCacheConfig.KAFKACACHE_ADMIN_CLIENT_TIMEOUT_MS_CONFIG] = 30000，避免网络波动时listTopics调用无限等待，同时添加cacheProps[KafkaCacheConfig.KAFKACACHE_RETRIES_CONFIG] = 3，允许缓存初始化失败自动重试，减少偶发超时导致的服务崩溃。
监控排查优化
给JVM添加堆内存dump参数：-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/存储路径/dump，如果后续再出现OOM可以直接分析dump文件，确认是数据量过大还是内存泄漏导致的问题。同时部署JVM监控指标，实时观测堆内存使用率、KCache条目数变化，提前进行扩容或者数据清理。

内容的提问来源于stack exchange，提问作者hermanjakobsen