[CAMEL-5526] "ThreadID is already used" exception under heavy load - ASF JIRA

Details

Type: Bug
Status: Resolved
Priority: Minor
Resolution: Fixed
Affects Version/s: 2.8.6, 2.9.2, 2.10.0, 2.11.0
Fix Version/s: 2.9.4, 2.10.2, 2.11.0
Component/s: camel-xmpp
Labels:
None

Patch Info:

Patch Available
Estimated Complexity:
Unknown

Description

In situations with heavy load or under load tests we rarely got exception:

 
2012-08-21 12:28:02,620 [ool-26-thread-7] ERROR DefaultErrorHandler            - Failed delivery for (MessageId: ID-atitov-pc-60956-1345537671897-2-143205 on ExchangeId: ID-atitov-pc-60956-1345537671897-2-143206). Exhausted after delivery attempt: 1 caught: java.lang.IllegalArgumentException: ThreadID is already used
java.lang.IllegalArgumentException: ThreadID is already used
	at org.jivesoftware.smack.ChatManager.createChat(ChatManager.java:163)
	at org.apache.camel.component.xmpp.XmppPrivateChatProducer.getOrCreateChat(XmppPrivateChatProducer.java:97)
	at org.apache.camel.component.xmpp.XmppPrivateChatProducer.process(XmppPrivateChatProducer.java:65)
	at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.SendProcessor$2.doInAsyncProducer(SendProcessor.java:120)
	at org.apache.camel.impl.ProducerCache.doInAsyncProducer(ProducerCache.java:292)
	at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:115)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.processor.interceptor.TraceInterceptor.process(TraceInterceptor.java:91)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.RedeliveryErrorHandler.processErrorHandler(RedeliveryErrorHandler.java:330)
	at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:220)
	at org.apache.camel.processor.RouteContextProcessor.processNext(RouteContextProcessor.java:45)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.processor.interceptor.DefaultChannel.process(DefaultChannel.java:303)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.Pipeline.process(Pipeline.java:117)
	at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)
	at org.apache.camel.processor.RouteContextProcessor.processNext(RouteContextProcessor.java:45)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.processor.UnitOfWorkProcessor.process(UnitOfWorkProcessor.java:122)
	at org.apache.camel.processor.RouteInflightRepositoryProcessor.processNext(RouteInflightRepositoryProcessor.java:48)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:73)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.component.direct.DirectProducer.process(DirectProducer.java:61)
	at org.apache.camel.processor.UnitOfWorkProcessor.processAsync(UnitOfWorkProcessor.java:150)
	at org.apache.camel.processor.UnitOfWorkProcessor.process(UnitOfWorkProcessor.java:117)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:99)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:86)
	at org.apache.camel.processor.UnitOfWorkProducer.process(UnitOfWorkProducer.java:63)
	at org.apache.camel.impl.ProducerCache$2.doInProducer(ProducerCache.java:360)
	at org.apache.camel.impl.ProducerCache$2.doInProducer(ProducerCache.java:1)
	at org.apache.camel.impl.ProducerCache.doInProducer(ProducerCache.java:227)
	at org.apache.camel.impl.ProducerCache.sendExchange(ProducerCache.java:331)
	at org.apache.camel.impl.ProducerCache.send(ProducerCache.java:169)
	at org.apache.camel.impl.DefaultProducerTemplate.send(DefaultProducerTemplate.java:111)
	at org.apache.camel.impl.DefaultProducerTemplate.sendBody(DefaultProducerTemplate.java:124)
	at org.apache.camel.impl.DefaultProducerTemplate.sendBody(DefaultProducerTemplate.java:131)
	at org.apache.camel.component.xmpp.XmppProducerConcurrentTest$1.call(XmppProducerConcurrentTest.java:79)
	at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
	at java.util.concurrent.FutureTask.run(FutureTask.java:138)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
	at java.lang.Thread.run(Thread.java:662)
2012-08-21 12:28:02,621 [ool-26-thread-5] ERROR DefaultErrorHandler            - Failed delivery for (MessageId: ID-atitov-pc-60956-1345537671897-2-143207 on ExchangeId: ID-atitov-pc-60956-1345537671897-2-143208). Exhausted after delivery attempt: 1 caught: java.lang.IllegalArgumentException: ThreadID is already used
java.lang.IllegalArgumentException: ThreadID is already used
	at org.jivesoftware.smack.ChatManager.createChat(ChatManager.java:163)
	at org.apache.camel.component.xmpp.XmppPrivateChatProducer.getOrCreateChat(XmppPrivateChatProducer.java:97)
	at org.apache.camel.component.xmpp.XmppPrivateChatProducer.process(XmppPrivateChatProducer.java:65)
	at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.SendProcessor$2.doInAsyncProducer(SendProcessor.java:120)
	at org.apache.camel.impl.ProducerCache.doInAsyncProducer(ProducerCache.java:292)
	at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:115)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.processor.interceptor.TraceInterceptor.process(TraceInterceptor.java:91)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.RedeliveryErrorHandler.processErrorHandler(RedeliveryErrorHandler.java:330)
	at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:220)
	at org.apache.camel.processor.RouteContextProcessor.processNext(RouteContextProcessor.java:45)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.processor.interceptor.DefaultChannel.process(DefaultChannel.java:303)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.Pipeline.process(Pipeline.java:117)
	at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)
	at org.apache.camel.processor.RouteContextProcessor.processNext(RouteContextProcessor.java:45)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.processor.UnitOfWorkProcessor.process(UnitOfWorkProcessor.java:122)
	at org.apache.camel.processor.RouteInflightRepositoryProcessor.processNext(RouteInflightRepositoryProcessor.java:48)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
	at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:73)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
	at org.apache.camel.component.direct.DirectProducer.process(DirectProducer.java:61)
	at org.apache.camel.processor.UnitOfWorkProcessor.processAsync(UnitOfWorkProcessor.java:150)
	at org.apache.camel.processor.UnitOfWorkProcessor.process(UnitOfWorkProcessor.java:117)
	at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:99)
	at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:86)
	at org.apache.camel.processor.UnitOfWorkProducer.process(UnitOfWorkProducer.java:63)
	at org.apache.camel.impl.ProducerCache$2.doInProducer(ProducerCache.java:360)
	at org.apache.camel.impl.ProducerCache$2.doInProducer(ProducerCache.java:1)
	at org.apache.camel.impl.ProducerCache.doInProducer(ProducerCache.java:227)
	at org.apache.camel.impl.ProducerCache.sendExchange(ProducerCache.java:331)
	at org.apache.camel.impl.ProducerCache.send(ProducerCache.java:169)
	at org.apache.camel.impl.DefaultProducerTemplate.send(DefaultProducerTemplate.java:111)
	at org.apache.camel.impl.DefaultProducerTemplate.sendBody(DefaultProducerTemplate.java:124)
	at org.apache.camel.impl.DefaultProducerTemplate.sendBody(DefaultProducerTemplate.java:131)
	at org.apache.camel.component.xmpp.XmppProducerConcurrentTest$1.call(XmppProducerConcurrentTest.java:79)
	at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
	at java.util.concurrent.FutureTask.run(FutureTask.java:138)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
	at java.lang.Thread.run(Thread.java:662)

After debugging and looking at source code we figured out, that smack use some sort of map with week references for chat cache. When memory is low java GC discards chat instances, but in Process() method they instantiated over and over again without synchronization:

XmppPrivateChatProducer.java

 public void process(Exchange exchange) {
 .........................
	Chat chat = chatManager.getThreadChat(endpoint.getChatId());
	if (chat == null) {
		LOG.trace("Creating new chat instance with thread ID {}", endpoint.getChatId());
		chat = chatManager.createChat(getParticipant(), endpoint.getChatId(), new MessageListener() {
			public void processMessage(Chat chat, Message message) {
				// not here to do conversation
				if (LOG.isDebugEnabled()) {
					LOG.debug("Received and discarding message from {} : {}", getParticipant(), message.getBody());
				}
			}
		});
	}
 .........................
}

But smack implementation prohibits chat instances with same chatId.

Attachments

- Sort By Name
- Sort By Date
- Ascending
- Descending

XmppPrivateChatProducer.patch
21/Aug/12 08:59
3 kB
Alexander Titov

"ThreadID is already used" exception under heavy load

Details

Description

Attachments

Attachments

Activity

People

Dates