[GitHub] [activemq-artemis] clebertsuconic commented on a change in pull request #3024: ARTEMIS-2664 The prefetch size is exceeded after delivered acks

[GitBox <gi...@apache.org>]

clebertsuconic commented on a change in pull request #3024: ARTEMIS-2664 The prefetch size is exceeded after delivered acks
URL: https://github.com/apache/activemq-artemis/pull/3024#discussion_r393872456
 
 

 ##########
 File path: tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/JmsClientAckTest.java
 ##########
 @@ -132,6 +137,72 @@ public void testUnAckedMessageAreNotConsumedOnSessionClose() throws JMSException
       session.close();
    }
 
+   /**
+    * Tests if acknowledged messages are being consumed.
+    *
+    * @throws JMSException
+    */
+   @Test
+   public void testAckedMessageDeliveringWithPrefetch() throws Exception {
+      final int prefetchSize = 10;
+      final int messageCount = 5 * prefetchSize;
+      connection.getPrefetchPolicy().setAll(prefetchSize);
+      connection.start();
+      Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+      Queue queue = session.createQueue(getQueueName());
+      QueueControl queueControl = (QueueControl)server.getManagementService().
+         getResource(ResourceNames.QUEUE + queueName);
+      MessageProducer producer = session.createProducer(queue);
+      for (int i = 0; i < messageCount; i++) {
+         producer.send(session.createTextMessage("MSG" + i));
+      }
+
+      // Consume the messages...
+      Message msg;
+      MessageConsumer consumer = session.createConsumer(queue);
+
+      Wait.assertEquals(0L, () -> queueControl.getMessagesAcknowledged(), 3000, 100);
+      Wait.assertEquals(prefetchSize, () -> queueControl.getDeliveringCount(), 3000, 100);
+
+      ArrayList<Message> messages = new ArrayList<>();
+      for (int i = 0; i < prefetchSize; i++) {
+         msg = consumer.receive(1000);
+         assertNotNull(msg);
+         messages.add(msg);
+      }
+
+      Wait.assertEquals(0L, () -> queueControl.getMessagesAcknowledged(), 3000, 100);
+      Wait.assertEquals(2 * prefetchSize, () -> queueControl.getDeliveringCount(), 3000, 100);
+
+      for (int i = 0; i < prefetchSize; i++) {
+         msg = messages.get(i);
+         msg.acknowledge();
+      }
+
+      Wait.assertEquals((long) prefetchSize, () -> queueControl.getMessagesAcknowledged(), 3000, 100);
+      Wait.assertEquals(prefetchSize, () -> queueControl.getDeliveringCount(), 3000, 100);
+
+      for (int i = 0; i < messageCount - prefetchSize; i++) {
+         msg = consumer.receive(1000);
+         assertNotNull(msg);
+         msg.acknowledge();
+      }
+
+      Wait.assertEquals((long)messageCount, () -> queueControl.getMessagesAcknowledged(), 3000, 100);
+      Wait.assertEquals(0, () -> queueControl.getDeliveringCount(), 3000, 100);
+
+      // Reset the session.
+      session.close();
+      session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+
+      // Attempt to Consume the message...
+      consumer = session.createConsumer(queue);
+      msg = consumer.receive(1000);
 
 Review comment:
   Please, don't use receive(timeout) if you expect to receive null... use receiveNoWait.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services