I found the setting, it's `maxPageSize`, see here: https://activemq.apache.org/per-destination-policies >From its description it's not clear to me that it also limits the number of msgs in a session.
Now another question: is it safe to set it to Integer.MAX_VALUE? What could be the consequences? Viliam On Wed, 18 Dec 2019 at 12:01, Viliam Durina <[email protected]> wrote: > 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]; >> } >> } >> } >> >
