当 num.stream.threads > 1 时出现零星的 ConcurrentModificationException答案

【问题标题】：Sporadic ConcurrentModificationException when num.stream.threads > 1当 num.stream.threads > 1 时出现零星的 ConcurrentModificationException
【发布时间】：2023-01-14 03:44:43
【问题描述】：

流偶尔会像这样遇到 ConcurrentModificationException：

[wrwks-bef-projekt-aggregat-wirtschaftseinheitAndMietobjektUpdater-2577dae3-7c43-4782-bdd8-e51669a18469-StreamThread-6] ERROR o.a.k.s.p.internals.TaskManager - stream-thread [wrwks-bef-projekt-aggregat-wirtschaftseinheitAndMietobjektUpdater-2577dae3-7c43-4782-bdd8-e51669a18469-StreamThread-6] Failed to process stream task 1_2 due to the following error:
org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=1_2, processor=KSTREAM-SOURCE-0000000011, topic=wrw-technischerplatz-mietobjekt-aggregat-oeffentlich-1, partition=2, offset=7263, stacktrace=java.util.ConcurrentModificationException
    at java.base/java.util.ArrayList$Itr.checkForComodification(Unknown Source)
    at java.base/java.util.ArrayList$Itr.next(Unknown Source)
    at org.apache.kafka.common.header.internals.RecordHeaders$FilterByKeyIterator.makeNext(RecordHeaders.java:184)
    at org.apache.kafka.common.header.internals.RecordHeaders$FilterByKeyIterator.makeNext(RecordHeaders.java:171)
    at org.apache.kafka.common.utils.AbstractIterator.maybeComputeNext(AbstractIterator.java:79)
    at org.apache.kafka.common.utils.AbstractIterator.hasNext(AbstractIterator.java:45)
    at org.springframework.cloud.stream.binder.kafka.streams.AbstractKafkaStreamsBinderProcessor$3.process(AbstractKafkaStreamsBinderProcessor.java:640)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:253)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:232)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:191)
    at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:42)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:253)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:232)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:191)
    at org.apache.kafka.streams.kstream.internals.KTableSource$KTableSourceProcessor.process(KTableSource.java:152)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:253)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:232)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:191)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:84)
    at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$1(StreamTask.java:731)
    at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:809)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:731)
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1296)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:784)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:604)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:576)

    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:758)
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1296)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:784)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:604)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:576)
Caused by: java.util.ConcurrentModificationException: null
    at java.base/java.util.ArrayList$Itr.checkForComodification(Unknown Source)
    at java.base/java.util.ArrayList$Itr.next(Unknown Source)
    at org.apache.kafka.common.header.internals.RecordHeaders$FilterByKeyIterator.makeNext(RecordHeaders.java:184)
    at org.apache.kafka.common.header.internals.RecordHeaders$FilterByKeyIterator.makeNext(RecordHeaders.java:171)
    at org.apache.kafka.common.utils.AbstractIterator.maybeComputeNext(AbstractIterator.java:79)
    at org.apache.kafka.common.utils.AbstractIterator.hasNext(AbstractIterator.java:45)
    at org.springframework.cloud.stream.binder.kafka.streams.AbstractKafkaStreamsBinderProcessor$3.process(AbstractKafkaStreamsBinderProcessor.java:640)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:253)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:232)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:191)
    at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:42)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:253)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:232)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:191)
    at org.apache.kafka.streams.kstream.internals.KTableSource$KTableSourceProcessor.process(KTableSource.java:152)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:253)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:232)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:191)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:84)
    at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$1(StreamTask.java:731)
    at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:809)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:731)
    ... 4 common frames omitted

我不确定这是 Kafka Streams 客户端还是应用程序代码中的错误：

@Configuration
class WirtschaftseinheitAndMietobjektUpdaterStreamConfiguration {
    @Bean
    fun wirtschaftseinheitAndMietobjektUpdater() =
        Function { projekte: KTable<String, ProjektAggregat> ->
            Function { mietobjekte: KTable<String, MietobjektAggregat> ->
                Function { wirtschaftseinheiten: KTable<String, WirtschaftseinheitAggregat> ->
                    projekte
                        .filter { _, projektAggregat -> projektAggregat.action == AGGREGATE }
                        .leftJoin(
                            mietobjekte,
                            { it.projekt?.projekt?.technischerPlatz },
                            { projektAggregat, mietobjekt ->
                                if (mietobjekt != null)
                                    if (mietobjekt.tplnr.length > WIRTSCHAFTSEINHEIT_LENGTH)
                                        (projektAggregat + mietobjekt).copy(action = MO_WE_UPDATE)
                                    else
                                        projektAggregat
                                else
                                    projektAggregat
                            },
                        )
                        .leftJoin(
                            wirtschaftseinheiten,
                            { it.projekt?.projekt?.technischerPlatz?.take(WIRTSCHAFTSEINHEIT_LENGTH) },
                            { projektAggregat, wirtschaftseinheit ->
                                if (wirtschaftseinheit != null)
                                    (projektAggregat + wirtschaftseinheit).copy(action = MO_WE_UPDATE)
                                else
                                    projektAggregat
                            },
                        )
                        .toStream()
                        .filter { _, projektAggregat -> projektAggregat?.action == MO_WE_UPDATE }
                        .transform({ EventTypeHeaderTransformer() })
                }
            }
        }
}

任何有用的提示将不胜感激。

【问题讨论】：

很难说，但您似乎正在访问记录标头，我认为问题出在 EventTypeHeaderTransformer 内？您的供应商是否在每次调用get() 时都返回一个新对象？
@MatthiasJ.Sax 是的，下面的测试是成功的： ` @Test fun supplierTest() { val x = TransformerSupplier { EventTypeHeaderTransformer() } val a = x.get() val b = x.get() assertNotEquals(a, b ）}`
好的。仍然不确定他的问题可能是什么。请注意，stacktrac 中有 org.springframework.cloud.stream.binder.kafka.streams.AbstractKafkaStreamsBinderProcessor$3.process(AbstractKafkaStreamsBinderProcessor.java:640)——也许 Spring 做错了什么？ -- 可能值得检查 Spring 代码......？

标签： apache-kafka spring-kafka apache-kafka-streams spring-cloud-stream-binder-kafka

【解决方案1】：

感谢@MatthiasJSax 的提示，我可以缩小问题范围并通过同步对 Kafka 标头的访问来解决它，这些标头可通过提供给转换器的上下文访问。

【讨论】：