JMSSink should cause CC to reset if message is not sent

[chanskw]

In the process method of JMSSink:

This is called if the message is not sent:

if (!msgSent) {
									
			logger.log(LogLevel.ERROR, "EXCEPTION_SINK");
			if (hasErrorPort) {
				sendOutputErrorMsg(tuple,
						"Dropping this tuple since an exception occurred while sending.");
			}
			
		}

If the JMSSink operator is in CC, we should trigger a reset in the CC.

[ghost]

In consistent region, the message is sent with sendMessageNoRetry:

else {
    // consistent region
    // Construct the JMS message based on the message type taking the
    // attributes from the tuple.
    // propagate all exceptions to the runtime to restart the consistent region in case of failure
    Message message = mhandler.convertTupleToMessage(tuple,
           jmsConnectionHelper.getSession());
    msgSent = jmsConnectionHelper.sendMessageNoRetry(message);
}

sendMessageNoRetry throws a JMSException in case of connection failure, which triggers reset of CR. Caught MessageFormatExceptions are NOT re-thrown, however:

boolean sendMessageNoRetry(Message message) throws JMSException {
	boolean res = false;
	try {
		// try to send the message
		synchronized (getSession()) {
			getProducer().send(message);
			res = true;
		}
	}
	catch (JMSException e) {
		// error has occurred, log error and try sending message again
		logger.log(LogLevel.WARN, "ERROR_DURING_SEND", new Object[] { e.toString() }); //$NON-NLS-1$
		// If the exception is caused by message format, then we can return peacefully as connection is still good.
/* ==>>  */	if(!(e instanceof MessageFormatException)) {
			throw e;
		}
	}
	if(!res) {
		nFailedInserts.incrementValue(1);
	}
	return res;
}

Thus, only in case of MessageFormatException the CR is not reset. Should also be reset. I suggest to throw also MessageFormatException, which is a subclass of JMSException and change the return of sendMessageNoRetryto void. Either we sent the message or we did not. If we did not, we throw an appropriate exception.

[schubon]

Check Rolef's comments and code changes. Probably, this issue may be closed already.