[jira] [Closed] (ARTEMIS-2293) addPacket method in the org.apache.activemq.artemis.core.client.impl.LargeMessageControllerImpl doesn't notify threads in case of an Exception

["Clebert Suconic (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/ARTEMIS-2293?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Clebert Suconic closed ARTEMIS-2293.
------------------------------------

> addPacket method in the org.apache.activemq.artemis.core.client.impl.LargeMessageControllerImpl  doesn't notify threads in case of an Exception
> -----------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: ARTEMIS-2293
>                 URL: https://issues.apache.org/jira/browse/ARTEMIS-2293
>             Project: ActiveMQ Artemis
>          Issue Type: Bug
>    Affects Versions: 2.19.0
>            Reporter: Pavel Molchanov
>            Priority: Major
>             Fix For: 2.20.0
>
>         Attachments: 13035609_ARTEMIS-2293-Test.patch
>
>          Time Spent: 1.5h
>  Remaining Estimate: 0h
>
> Block that handles exceptions in the catch(Exception e) doesn't call notifyAll(). That cause that other working threads are not released in the waitCompletion method.
> [https://github.com/apache/activemq-artemis/blob/master/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java]
>  
> addPacket method:
> {code:java}
> public void addPacket(byte[] chunk, int flowControlSize, boolean isContinues) {
> 	int flowControlCredit = 0;
> 	
> 	synchronized (this) {
> 	packetAdded = true;
> 	if (outStream != null) {
> 	try {
> 	if (!isContinues) {
> 	streamEnded = true;
> 	}
> 	
> 	if (fileCache != null) {
> 	fileCache.cachePackage(chunk);
> 	}
> 	
> 	outStream.write(chunk);
> 	
> 	flowControlCredit = flowControlSize;
> 	
> 	notifyAll();
> 	
> 	if (streamEnded) {
> 	outStream.close();
> 	}
> 	} catch (Exception e) {
> 	ActiveMQClientLogger.LOGGER.errorAddingPacket(e);
> 	handledException = e;
> 	}
> 	} else {
> 	if (fileCache != null) {
> 	try {
> 	fileCache.cachePackage(chunk);
> 	} catch (Exception e) {
> 	ActiveMQClientLogger.LOGGER.errorAddingPacket(e);
> 	handledException = e;
> 	}
> 	}
> 	
> 	largeMessageData.offer(new LargeData(chunk, flowControlSize, isContinues));
> 	}
> 	}{code}
>  
> waitCompletion method:
> {code:java}
> public synchronized boolean waitCompletion(final long timeWait) throws ActiveMQException {
> 	if (outStream == null) {
> 	// There is no stream.. it will never achieve the end of streaming
> 	return false;
> 	}
> 	
> 	long timeOut;
> 	
> 	// If timeWait = 0, we will use the readTimeout
> 	// And we will check if no packets have arrived within readTimeout milliseconds
> 	if (timeWait != 0) {
> 	timeOut = System.currentTimeMillis() + timeWait;
> 	} else {
> 	timeOut = System.currentTimeMillis() + readTimeout;
> 	}
> 	
> 	while (!streamEnded && handledException == null) {
> 	try {
> 	this.wait(timeWait == 0 ? readTimeout : timeWait);
> 	} catch (InterruptedException e) {
> 	throw new ActiveMQInterruptedException(e);
> 	}
> 	
> 	if (!streamEnded && handledException == null) {
> 	if (timeWait != 0 && System.currentTimeMillis() > timeOut) {
> 	throw ActiveMQClientMessageBundle.BUNDLE.timeoutOnLargeMessage();
> 	} else if (System.currentTimeMillis() > timeOut && !packetAdded) {
> 	throw ActiveMQClientMessageBundle.BUNDLE.timeoutOnLargeMessage();
> 	}
> 	}
> 	}
> 	
> 	checkException();
> 	
> 	return streamEnded;
> 	
> 	}{code}
>  
>  



--
This message was sent by Atlassian Jira
(v8.20.1#820001)