Just found out this is also the case for non-XA transactions or in CLIENT_ACKNOWLEDGE mode: only up to 200 messages are received. I'm working on a stream processing engine and we need to consume messages for certain time and acknowledge them afterwards. I couldn't find a config to increase this value or set it to unbounded.
So the question is: is there a way to increase the limit of unacknowledged messages in a session? Viliam On Wed, 4 Dec 2019 at 16:12, Viliam Durina <[email protected]> wrote: > Hi all, > > I'm trying to use the XA transactions. I produce 1k items into a queue > using an auto-ack session. Then I try to consume them in an XA transaction, > but only 200 items are received. > > The output of the following test is: > java.lang.AssertionError: > Expected :1000 > Actual :200 > > Why is that? Is that expected? > > Viliam > > -- Test code: > > public class JmsXaTest { > @ClassRule > public static EmbeddedActiveMQBroker broker = new > EmbeddedActiveMQBroker(); > private XAConnectionFactory cf = new > ActiveMQXAConnectionFactory(broker.getVmURL()); > > @Test > public void test() throws Exception { > // produce 1k items to the queue > try ( > Connection conn = ((ConnectionFactory) cf).createConnection(); > Session session = conn.createSession(false, AUTO_ACKNOWLEDGE); > MessageProducer producer = > session.createProducer(session.createQueue("queue")) > ) { > for (int i = 0; i < 1_000; i++) { > producer.send(session.createTextMessage("msg-" + i)); > } > } > > // try to consume all items in an XA transaction > XAConnection conn = cf.createXAConnection(); > conn.start(); > XASession sess = conn.createXASession(); > Xid xid1 = new MyXid(1); > sess.getXAResource().start(xid1, XAResource.TMNOFLAGS); > MessageConsumer cons1 = > sess.createConsumer(sess.createQueue("queue")); > int count = 0; > for (; cons1.receive(3000) != null; count++) { > System.out.println(count); > } > assertEquals(1_000, count); > } > > private static class MyXid implements Xid { > private byte[] gtrid; > > public MyXid(int val) { > gtrid = new byte[]{(byte) (val)}; > } > > @Override > public int getFormatId() { > return 1; > } > > @Override > public byte[] getGlobalTransactionId() { > return gtrid; > } > > @Override > public byte[] getBranchQualifier() { > return new byte[1]; > } > } > } >
