ConcurrentModificationException in DefaultSubscriptionRegistry cache [SPR-11755]

[spring-projects-issues]

Sébastien Deleuze opened SPR-11755 and commented

When a subscription in unregistered with remaining subscriptions in the cache, a ConcurrentModificationException is thrown by the DestinationCache class.

Exception in thread "clientInboundChannel-1" Exception in thread "clientInboundChannel-2" java.util.ConcurrentModificationException
	at java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:711)
	at java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:734)
	at org.springframework.messaging.simp.broker.DefaultSubscriptionRegistry$DestinationCache.updateAfterRemovedSubscription(DefaultSubscriptionRegistry.java:191)
	at org.springframework.messaging.simp.broker.DefaultSubscriptionRegistry.removeSubscriptionInternal(DefaultSubscriptionRegistry.java:100)
	at org.springframework.messaging.simp.broker.AbstractSubscriptionRegistry.unregisterSubscription(AbstractSubscriptionRegistry.java:91)
	at org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler.handleMessageInternal(SimpleBrokerMessageHandler.java:129)
	at org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler.handleMessage(AbstractBrokerMessageHandler.java:171)
	at org.springframework.messaging.support.ExecutorSubscribableChannel$1.run(ExecutorSubscribableChannel.java:70)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:744)
java.util.ConcurrentModificationException
	at java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:711)
	at java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:734)
	at org.springframework.messaging.simp.broker.DefaultSubscriptionRegistry$DestinationCache.updateAfterRemovedSubscription(DefaultSubscriptionRegistry.java:191)
	at org.springframework.messaging.simp.broker.DefaultSubscriptionRegistry.removeSubscriptionInternal(DefaultSubscriptionRegistry.java:100)
	at org.springframework.messaging.simp.broker.AbstractSubscriptionRegistry.unregisterSubscription(AbstractSubscriptionRegistry.java:91)
	at org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler.handleMessageInternal(SimpleBrokerMessageHandler.java:129)
	at org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler.handleMessage(AbstractBrokerMessageHandler.java:171)
	at org.springframework.messaging.support.ExecutorSubscribableChannel$1.run(ExecutorSubscribableChannel.java:70)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:744)
}}

The cause seems to be updateCache modification during iterating over its elements rather that concurrent access by 2 threads (updateCache modification are already synchronized).

It can be fixed in DestinationCache by :

• Moving the cached destinations removal outside the for loop.
• Using updateCache.entrySet() instead of updateCache.keySet() + updateCache.get() in order to avoid updateCache modification during iteration. The tricky point here is we are using a LinkedHashMap with accessOrder=true. A simple updateCache.get() modify the map. By using updateCache.entrySet() we avoid this update.

---

Affects: 4.0.4

Referenced from: commits 96da77e, 98738c0