ConcurrentModificationException from ThreadStatePropagationChannelInterceptor

[gautham-kishtapuram]

This issue has been observed with Spring Integration Core version 6.3.0.

Detailed Versioning Summary

• Before Upgrade:
  • Java version: 17
  • Spring Boot version: 3.0.1
  • spring-boot-starter-integration: 3.0.1
  • spring-integration-core: 6.0.1
• After Upgrade:
  • Java version: 21
  • Spring Boot version: 3.3.0
  • spring-boot-starter-integration: 3.3.0
  • spring-integration-core: 6.3.0

---

Background and Problem Details
Our project was updated from Java 17 to Java 21 and from spring-integration-core:6.0.1 to 6.3.0.

So in our Spring Integration setup, we implemented a custom ThreadStatePropagationChannelInterceptor to manage MDC (Mapped Diagnostic Context) propagation across asynchronous message channels. However, after the update -- especically spring-integration-core:6.3.0, we began encountering ConcurrentModificationException errors during message processing.Additionally, this only occurs when the application is in debug mode.

Exception Stack Trace

   Caused by: java.util.ConcurrentModificationException
  at java.base/java.util.LinkedList$ListItr.checkForComodification(LinkedList.java:977) ~[?:?]
  at java.base/java.util.LinkedList$ListItr.next(LinkedList.java:899) ~[?:?]
  at java.base/java.util.AbstractCollection.toString(AbstractCollection.java:458) ~[?:?]
  at java.base/java.lang.StringConcatHelper.stringOf(StringConcatHelper.java:467) ~[?:?]
  at org.springframework.integration.channel.interceptor.ThreadStatePropagationChannelInterceptor$MessageWithThreadState.toString(ThreadStatePropagationChannelInterceptor.java:134) ~[spring-integration-core-6.3.0.jar!/:6.3.0]
  at java.base/java.lang.StringConcatHelper.stringOf(StringConcatHelper.java:467) ~[?:?]
  at org.springframework.integration.channel.AbstractMessageChannel.sendInternal(AbstractMessageChannel.java:381) ~[spring-integration-core-6.3.0.jar!/:6.3.0]
  at org.springframework.integration.channel.AbstractMessageChannel.sendWithMetrics(AbstractMessageChannel.java:349) ~[spring-integration-core-6.3.0.jar!/:6.3.0]
  at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:329) ~[spring-integration-core-6.3.0.jar!/:6.3.0]
  at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:302) ~[spring-integration-core-6.3.0.jar!/:6.3.0]
  

Screenshot from where this expection is being raised

This Screen shot shows that when the application runs in debug mode, a toString method on **messageToSend** is invoked. where, messageToSend contains an implementation of ThreadStatePropagationInterceptor

---

Analysis :

According to this PR: https://github.com/Fix ThreadSPropagationChInterceptor for stacking #8735, an update to ThreadStatePropagationChannelInterceptor introduced a design change that stores each context state in stateQueue within MessageWithThreadState.

This issue appears to stem from concurrent access to ThreadStatePropagationChannelInterceptor.MessageWithThreadState.stateQueue, which is currently implemented as a LinkedList and is not thread-safe.

Because In this new approach, the postReceive() method calls poll() on the stateQueue to retrieve and clear the oldest context while preserving the order of interceptors. However, this fix appears to have introduced a concurrency issue. Since the poll() method modifies the **stateQueue** object, which can lead to a ConcurrentModificationException when multiple threads access stateQueue simultaneously. In my case, I observed that while one thread performs a toString operation on a LinkedList, another thread calls poll(), resulting in this CMException

---

To Reproduce
Ensure your project is using Spring Integration Core version 6.3.0.
Implement a custom ThreadStatePropagationChannelInterceptor for context propagation.
Run the application in debug mode and send messages through an integration flow that routes messages to couple of asynchronous channels. Monitor for any ConcurrentModificationException during message processing.

Expected behavior
Concurrent access to stateQueue should not result in a ConcurrentModificationException. Ideally, stateQueue would be thread-safe to support multi-threaded access during message processing.

[artembilan]

I think I'll fix it use a LinkedBlockingQueue instead.

As a workaround I suggest to look into a solution with a message header.
See SecurityContextPropagationChannelInterceptor, for example.

And I mentioned that in the PR you've linked here: #8735 (comment)