Error while Creating JMS Message Producer in jboss EAP 7

Question

I have configured JMS topic in JBOSS_EAP_7.0 and write a simple java code to create a message producer. I have the following stateless bean

@Stateless
public class ExchangeSenderFacadeWrapperBean {


    private static final OMSLogHandlerI logger = new Log4j2Handler("ClientSenderFacadeBean");
    @Resource(lookup = "java:/JmsXA")     // inject ConnectionFactory (more)
    protected ConnectionFactory  factory;


    @Resource(lookup = "java:/jms/topic/ORD_CLINT_PUSH")
    protected Topic target;

    private Connection  connection = null;
    private Session session = null;



    public void sendMessage(String message) {

        MessageProducer producer= null;
        try {
            if(connection==null){  //todo verify
                connection = factory.createConnection();
            }
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            producer = session.createProducer(target);
            producer.setDisableMessageID(true);
            TextMessage outmsg = session.createTextMessage(message);
            producer.send(outmsg);
            logger.info("Message was sent to Topic");
            producer.setTimeToLive(900000);//15min  //todo
        } catch (Exception e) {
            logger.error(" Error when sending order to jboss:", e);
            throw new OMSCoreRuntimeException(e.getMessage(), e);
        } finally {
            try {
                if (producer != null)
                    producer.close();
            } catch (JMSException e) {
                logger.warn("\n jms producer close error:",e);
            }
            try {
                if (session != null)
                    session.close();
            } catch (JMSException e) {
                logger.warn("\n jms session close error:",e);
            }
        }
    }

This works fine until i made simple change to move sendMessage(String message) method to pojo class as follow.

@Stateless(name = "ExchangeSenderFacadeBean")
@Local({ExchangeSenderFacadeLocalI.class})
public class ExchangeSenderFacadeWrapperBean implements ExchangeSenderFacadeLocalI {
    @Resource(lookup = "java:/JmsXA")     // inject ConnectionFactory (more)
    protected ConnectionFactory factory;

    @EJB(beanName = "BeanRegistryLoader")
    protected BeanRegistryLoader omsRegistryBean;

    protected BeanRegistryCore beanRegistryCore;

    @Resource(lookup = "java:/jms/queue/ToExchange")
    protected Queue target;

    private ExchangeSenderFacadeCoreI exchangeSenderFacadeCore;


    @Override
    public void sendToExchange(ExchangeMessage exchangeMessage) {
        exchangeSenderFacadeCore.sendToExchange(exchangeMessage);

    }

    @PostConstruct
    public void init() {
        beanRegistryCore = omsRegistryBean.registry();
        if (exchangeSenderFacadeCore == null) {
            exchangeSenderFacadeCore = ((BeanRegistryCore) omsRegistryBean.registry()).getExchangeSenderFacadeCoreI();
            exchangeSenderFacadeCore.setBeanRegistryCore(omsRegistryBean.registry());
            exchangeSenderFacadeCore.setFactory(factory);
            exchangeSenderFacadeCore.setTargetQueue(target);
        }
    }

}

ConnectionFactory and target Queue variables set inside EJB PostConstruct method and pojo class looks like follow, which now contains logic to create and publish method to EJB queue

public class ExchangeSenderFacadeCore implements ExchangeSenderFacadeCoreI {
    private static final OMSLogHandlerI logger = new Log4j2HndlAdaptor("ExchangeSenderFacadeCore");
    private BeanRegistryCore beanRegistryCore;
    private ConnectionFactory factory;
    private Connection connection = null;
    private Session session = null;
    private long ttl = 900000;
    protected Queue targetQueue;

    public ExchangeSenderFacadeCore() {
        if (System.getProperty(OMSConst.SYS_PROPERTY_JMS_TTL) != null && System.getProperty(OMSConst.SYS_PROPERTY_JMS_TTL).length() > 0) {
            ttl = Long.parseLong(System.getProperty(OMSConst.SYS_PROPERTY_JMS_TTL));
        }
        logger.info("LN:103", "==JMS Topic TTL:" + ttl);
    }

    @Override
    public void processSendToExchange(ExchangeMessage exchangeMessage) {
        sendToExchange(exchangeMessage);
    }

    public boolean isParallelRunEnabled() {
        Object isParallelRun = beanRegistryCore.getCacheAdaptorI().cacheGet(OMSConst.DEFAULT_TENANCY_CODE, OMSConst.APP_PARAM_IS_PARALLEL_RUN, CACHE_NAMES.SYS_PARAMS_CACHE_CORE);
        if (isParallelRun != null && String.valueOf(isParallelRun).equals(OMSConst.STRING_1)) {
            return true;
        }
        return false;
    }

    @Override
    public void sendToExchange(ExchangeMessage exchangeMessage) {
        MessageProducer producer = null;
        try {
            if (isParallelRunEnabled()) {
                logger.info("LN:66", "== Message send to exchange skipped,due to parallel run enabled");
                return;
            }
            if (connection == null) {
                connection = factory.createConnection();
            }
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            producer = session.createProducer(targetQueue);
            producer.setDisableMessageID(true);
            Message message = beanRegistryCore.getJmsExchangeMsgTransformerI().transformToJMSMessage(session, exchangeMessage);
            producer.send(message);
            producer.setTimeToLive(ttl);//default 15min
            logger.elkLog("78", "-1", LogEventsEnum.SENT_TO_EXCHANGE, exchangeMessage.toString());
        } catch (Exception e) {
            logger.error("LN:80", " Error when sending order to exchange:", e);
            throw new OMSCoreRuntimeException(e.getMessage(), e);
        } finally {
            try {
                if (producer != null)
                    producer.close();
            } catch (JMSException e) {
                logger.error("LN:87", "JMS producer close error:", e);
            }
            try {
                if (session != null)
                    session.close();
            } catch (JMSException e) {
                logger.error("LN:93", "JMS session close error:", e);
            }
        }
    }

    @Override
    public void processSendToExchangeSync(ExchangeMessage exchangeMessage) {

    }

    @Override
    public BeanRegistryCore getBeanRegistryCore() {
        return beanRegistryCore;
    }

    @Override
    public void setBeanRegistryCore(BeanRegistryCore beanRegistryCore) {
        this.beanRegistryCore = beanRegistryCore;
    }

    @Override
    public ConnectionFactory getFactory() {
        return factory;
    }

    @Override
    public void setFactory(ConnectionFactory factory) {
        this.factory = factory;
    }

    @Override
    public Queue getTargetQueue() {
        return targetQueue;
    }

    @Override
    public void setTargetQueue(Queue targetQueue) {
        this.targetQueue = targetQueue;
    }
}

But when i execute moderated code it gives me following error

javax.ejb.EJBTransactionRolledbackException: Producer is closed

Any possible fixes?

Can you provide the full stack-trace for the `EJBTransactionRolledbackException`? — Justin Bertram, Aug 14 '19 at 15:09
Also, there's no real need to cache the `connection` since you're getting it from the `JmsXA` connection factory which is pooled. You can simply create/close the connection every time. Of course, if the connection factory wasn't pooled this would be an anti-pattern. — Justin Bertram, Aug 14 '19 at 15:11
@JustinBertram To back your argument I get the following exception too. ```javax.ejb.EJBTransactionRolledbackException: Only allowed one session per connection``` When I create/close the connection every time as you mentioned I did not get an exception and works fine but the concern is even by caching connection message producer works fine in scenario 1. It only fails ever since I move ```sendMessage()``` method to POJO class as in Scenario 2. Any Logic behind this scene ?? — IsharaD, Aug 14 '19 at 15:35
If it works when you create/close the connection every time then I would go with that. This simplifies your code and allows the pool to do what it was designed to do. When you cache connections from a pool especially in an environment like Java EE where there are particular rules regarding what you are allowed to do with a connection (e.g. only one session per connection) then weird things can happen. I can't really offer much more than that without an example I can actually run to see what's happening. — Justin Bertram, Aug 14 '19 at 17:04
@JustinBertram Thank you for all your help in this case and I did some research and came with the following answer could you please verify it? It would be a great help for me. — IsharaD, Aug 18 '19 at 17:01

score 1 · Accepted Answer · answered Aug 18 '19 at 16:59

After a deep search deep into the problem, I found this https://developer.jboss.org/wiki/ShouldICacheJMSConnectionsAndJMSSessions article posted on one of JBOSS developer thread. This explains clearly the reason for caching connection and other JMS related resources being an anti-pattern for JMS code is running in a JEE application server.

In nutshell JCA layer pools JMS connections and JMS sessions. So when you call createConnection() or createSession(), then, in most cases it's not really calling the actual JMS implementation to actually create a new JMS connection or JMS session, it's just returning one from its own internal cache.

Additionally JBOSS server too manages stateless session bean pool. A stateless session bean is available on the connection pool only after you are done with the purpose of it, but not prior. Meantime Connection (Either JMS newly created or Cached) used to create JMS Session (session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) inside stateless session bean, also done with its purpose and too available on JCA layer connection pool. Therefore calling cached connection inside stateless EJB class as follow will not give you an exception even though it is not recommended by Oracle.

public void sendToExchange(ExchangeMessage exchangeMessage) {
        MessageProducer producer = null;
        try {
            if (isParallelRunEnabled()) {
                logger.info("LN:66", "== Message send to exchange skipped,due to parallel run enabled");
                return;
            }
            if (connection == null) {
                connection = factory.createConnection();
            }
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            producer = session.createProducer(targetQueue);
            producer.setDisableMessageID(true);
            Message message = beanRegistryCore.getJmsExchangeMsgTransformerI().transformToJMSMessage(session, exchangeMessage);
            producer.send(message);
            producer.setTimeToLive(ttl);//default 15min
            logger.elkLog("78", "-1", LogEventsEnum.SENT_TO_EXCHANGE, exchangeMessage.toString());
        } catch (Exception e) {
            logger.error("LN:80", " Error when sending order to exchange:", e);
            throw new OMSCoreRuntimeException(e.getMessage(), e);
        } finally {
            try {
                if (producer != null)
                    producer.close();
            } catch (JMSException e) {
                logger.error("LN:87", "JMS producer close error:", e);
            }
            try {
                if (session != null)
                    session.close();
            } catch (JMSException e) {
                logger.error("LN:93", "JMS session close error:", e);
            }
        }
    }

But in this case, since the same POJO class instance can be used on multiple occasions as bellow. It does not guarantee that the connection is freed and available in the JCA layer connection pool and gives exceptions.

@PostConstruct
    public void init() {
        beanRegistryCore = omsRegistryBean.registry();
        if (exchangeSenderFacadeCore == null) {
            exchangeSenderFacadeCore = ((BeanRegistryCore) omsRegistryBean.registry()).getExchangeSenderFacadeCoreI();
            exchangeSenderFacadeCore.setBeanRegistryCore(omsRegistryBean.registry());
            exchangeSenderFacadeCore.setFactory(factory);
            exchangeSenderFacadeCore.setTargetQueue(target);
        }
    }

Error while Creating JMS Message Producer in jboss EAP 7

1 Answers1