Limited XA transaction size?

[Viliam Durina]

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];
        }
    }
}