Issue

A simple test case with a AMQP consumer using a transacted JMS session session = con.createSession(true,Session.AUTO_ACKNOWLEDGE); shows that the consumer is not acknowledging message consumption properly. Hence, the messages continue to remain on the queue even after the consumer has processed all of them.

The consumer does issue session.commit() after processing the messages. However, it does not seem to direct the broker to remove the messages from the queue.

Code snippet is below.

        try{
            connectionFactory =
                new ConnectionFactoryImpl(hostName,port,userName,password,clientId,hostName,false);
            con = connectionFactory.createConnection();
            session = con.createSession(true,Session.AUTO_ACKNOWLEDGE);
            con.start();
            queue = session.createQueue(destinationName);
            String msgSelector="type = '1'";
            mc = session.createConsumer(queue,msgSelector);
            for(int i=0;i<10;i++){
                TextMessage tm = (TextMessage)mc.receive(1000);
                System.out.println("Text message Received " + tm.getText());
            }
            session.commit();
        }catch(Exception e){
            e.printStackTrace();
        }finally{
            try{
                if(session != null){
                    session.close();
                }
                if(mc != null){
                    mc.close();
                }
                if (connectionFactory != null){
                        con.stop();
                        con.close();
                }
            }catch(Exception jmsE){
                jmsE.printStackTrace();
            }

        }