You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@activemq.apache.org by "Justin Bertram (Jira)" <ji...@apache.org> on 2021/10/25 17:11:00 UTC
[jira] [Updated] (ARTEMIS-3110) PullMessage timeout is ignored
[ https://issues.apache.org/jira/browse/ARTEMIS-3110?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Justin Bertram updated ARTEMIS-3110:
------------------------------------
Component/s: OpenWire
> PullMessage timeout is ignored
> ------------------------------
>
> Key: ARTEMIS-3110
> URL: https://issues.apache.org/jira/browse/ARTEMIS-3110
> Project: ActiveMQ Artemis
> Issue Type: Bug
> Components: OpenWire
> Affects Versions: 2.16.0
> Reporter: Svetlana Undalova
> Priority: Major
>
> I ran into an issue trying to use Apache Artemis (2.16).
> I have slow consumers and went with prefetch size = 0 which according to docs results in pulling messages by the client. All works well if consumer starts after messages are already in the queue, but if it becomes idle after a while and manages to call consumer.Receive() N times then next time when messages appear in the queue I can see that all N are sent to the client (via checking UnconsumedMessageCount)
> I checked the same set up with ActiveMQ (5.16) and there both scenarios work correctly. I think that issue may happen because Artemis broker ignores PullMessage timeout.
> Here is my code
>
> {code:java}
> static async Task Main(string[] args)
> {
> var connectionfactory = new Apache.NMS.ActiveMQ.ConnectionFactory("tcp://localhost:61616?connection.PrefetchPolicy.queuePrefetch=0");
>
> var slowConsumer = ConsumerProcess("slow consumer", TimeSpan.FromSeconds(120), connectionfactory);
> var fastConsumer = ConsumerProcess("fast consumer", TimeSpan.FromSeconds(10), connectionfactory);
>
> // To make sure several pull requests are sent
> await Task.Delay(TimeSpan.FromSeconds(5));
>
> // Adding messages
> using (var connection = connectionfactory.CreateConnection("user", "password")) {
> connection.Start();
> using (var session = connection.CreateSession(acknowledgementMode: Apache.NMS.AcknowledgementMode.AutoAcknowledge))
> {
> var destination = Apache.NMS.Util.SessionUtil.GetDestination(session, "queue://cookies");
> using (var producer = session.CreateProducer(destination)) {
> for (int i = 0; i < 10; i++)
> {
> var message = producer.CreateTextMessage(); message.Text = $"{i}_{DateTimeOffset.Now}_{Guid.NewGuid()}";
> Console.WriteLine($"{DateTimeOffset.Now}\tSending message {message.Text}");
> producer.Send(message);
> }
> }
> }
> }
> Console.WriteLine("Press enter to exit");
> Console.ReadLine();
> }
> private static async Task ConsumerProcess(string name, TimeSpan processingTime, Apache.NMS.IConnectionFactory connectionfactory)
> {
> using (var connection = connectionfactory.CreateConnection("user", "password")) {
> connection.Start();
> using (var session = connection.CreateSession(acknowledgementMode: Apache.NMS.AcknowledgementMode.IndividualAcknowledge))
> {
> var destination = Apache.NMS.Util.SessionUtil.GetDestination(session, "queue://cookies");
> using (var consumer = session.CreateConsumer(destination)) {
> while (true)
> {
> var message = consumer.Receive(TimeSpan.FromSeconds(1));
> if (message is Apache.NMS.ITextMessage textMessage)
> {
> Console.WriteLine($"{DateTimeOffset.Now}\t{name} received message - {textMessage.Text}");
> Console.WriteLine($"{DateTimeOffset.Now}\t{name} unconsumed messages after receive - {((Apache.NMS.ActiveMQ.MessageConsumer)consumer).UnconsumedMessageCount}"); await Task.Delay(processingTime);
> message.Acknowledge();
> }
> else
> {
> await Task.Delay(TimeSpan.FromSeconds(1));
> }
> }
> }
> }
> }
> }
> {code}
>
>
> Here is output
> {code:java}
> 05.02.2021 16:04:55 +01:00 Sending message 5_05.02.2021 16:04:55 +01:00_97a6857b-d003-4c63-ba11-16a6825e6595
> 05.02.2021 16:04:55 +01:00 Sending message 6_05.02.2021 16:04:55 +01:00_4581b335-2483-42ac-a9c5-fe530a870ee1
> 05.02.2021 16:04:55 +01:00 Sending message 7_05.02.2021 16:04:55 +01:00_2ad2c9d4-aab6-47be-b51c-a1faacd6a03c
> 05.02.2021 16:04:55 +01:00 Sending message 8_05.02.2021 16:04:55 +01:00_866bca6b-9237-491b-981f-d73ae7c025fb
> 05.02.2021 16:04:55 +01:00 Sending message 9_05.02.2021 16:04:55 +01:00_a60b926c-96b2-4989-8f5f-ec85e5cfb071
> Press enter to exit
> 05.02.2021 16:04:56 +01:00 slow consumer received message - 0_05.02.2021 16:04:55 +01:00_ba96ec80-3008-4c95-987c-5f82a1aaa239
> 05.02.2021 16:04:56 +01:00 slow consumer unconsumed messages after receive - 3
> 05.02.2021 16:05:06 +01:00 fast consumer received message - 3_05.02.2021 16:04:55 +01:00_34df251c-da75-4145-8da4-efeecea3318d
> 05.02.2021 16:05:06 +01:00 fast consumer unconsumed messages after receive - 2
> 05.02.2021 16:05:17 +01:00 fast consumer received message - 5_05.02.2021 16:04:55 +01:00_97a6857b-d003-4c63-ba11-16a6825e6595
> 05.02.2021 16:05:17 +01:00 fast consumer unconsumed messages after receive - 1
> 05.02.2021 16:05:28 +01:00 fast consumer received message - 7_05.02.2021 16:04:55 +01:00_2ad2c9d4-aab6-47be-b51c-a1faacd6a03c
> 05.02.2021 16:05:28 +01:00 fast consumer unconsumed messages after receive - 0
> 05.02.2021 16:05:39 +01:00 fast consumer received message - 8_05.02.2021 16:04:55 +01:00_866bca6b-9237-491b-981f-d73ae7c025fb
> 05.02.2021 16:05:39 +01:00 fast consumer unconsumed messages after receive - 0
> 05.02.2021 16:05:50 +01:00 fast consumer received message - 9_05.02.2021 16:04:55 +01:00_a60b926c-96b2-4989-8f5f-ec85e5cfb071
> 05.02.2021 16:05:50 +01:00 fast consumer unconsumed messages after receive - 0
> 05.02.2021 16:06:57 +01:00 slow consumer received message - 2_05.02.2021 16:04:55 +01:00_86962504-1b4b-4f2e-b359-585e535042e9
> 05.02.2021 16:06:57 +01:00 slow consumer unconsumed messages after receive - 2
> 05.02.2021 16:08:58 +01:00 slow consumer received message - 4_05.02.2021 16:04:55 +01:00_2d7ce61f-0240-4278-a779-936fd4f3a3e7
> 05.02.2021 16:08:58 +01:00 slow consumer unconsumed messages after receive - 1
> 05.02.2021 16:10:59 +01:00 slow consumer received message - 6_05.02.2021 16:04:55 +01:00_4581b335-2483-42ac-a9c5-fe530a870ee1
> 05.02.2021 16:10:59 +01:00 slow consumer unconsumed messages after receive - 0
>
> {code}
> As you may see consumers received more than 1 message (according to unconsumed message count which is amount of messages that were sent to the client). I was playing with it and usually amount of messages received are equals to amount of Receive() calls I made.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)