I'm sorry, that was a typo. It should have read from_server_hb_millis = 0,
which is the way I ran it, and the way I plan to run it in the future.
I have very little control over how long it will take to process a message --
sometimes it can take hours.
I just wanted to confirm that this was expected behaviour and to give anyone
else with a similar use case a head up.
At the end of the day, and after two weeks of work, I finally have a working
solution, i.e., I have successfully migrated from Classic to Artemis.
And all of this just so I can leverage more granular priorites!
________________________________
From: Robbie Gemmell <robbie.gemm...@gmail.com>
Sent: Friday, June 24, 2022 10:50 AM
To: users@activemq.apache.org <users@activemq.apache.org>
Subject: Re: [External] - Re: How to avoid AMQ229014 TTL "dump" for a "patient"
Python 3 Stomp Listener
CAUTION: This email originated from outside of the organization. Do not click
links or open attachments unless you recognize the sender and know the content
is safe.
Both your 'as is' code and 'on the other hand' description have
from_server_hb_millis = 10000 so its not clear what the other one was
meant to be. Since you later referenced 'not requesting server HBs'
I'd maybe guess your working case was just leaving it at the default
value of 0. Presumably setting it to a higher value, larger than your
expected message processing time, would also be workable.
On Fri, 24 Jun 2022 at 14:30, Richard Bergmann
<rbergm...@colsa.com.invalid> wrote:
>
> Sorry, I hope I am not beating a dead horse here, but I thought it might be
> helpful to report the results of my investigation.
>
> First of all, recall that my use case is the listener could be patient for
> long periods of time but then become busy on a long-running task as the
> result of a message received -- much longer than TTL in both cases.
>
> Now consider this code, which I used in my investigation:
>
> -------
> # Message Consumer/Worker that monitors the "AAA" queue.
> from json import dumps
> import loggers
> from stomp import (Connection12,
> ConnectionListener)
> import sys
> import time
>
>
> logger = loggers.newlogger(__name__)
>
>
> # Listener that consumes messages requesting AAA processing.
> class AAAHeartbeatListener(ConnectionListener):
> def __init__(self,
> conn):
> self.conn = conn
>
> def on_connected(self,
> frame):
> # Called by the STOMP connection when a CONNECTED frame is received
> (after a connection has been established or re-established).
> # Parameters: frame (Frame) - the stomp frame
> logger.info("on_connected: " + str(frame))
>
> def on_connecting(self,
> host_and_port):
> # Called by the STOMP connection once a TCP/IP connection to the
> STOMP server has been established or re-established. Note that at this point,
> no connection has been established on the STOMP protocol level. For this, you
> need to invoke the “connect" method on the connection.
> # Parameters: host_and_port ((str,int)) - a tuple containing the host
> name and port number to which the connection has been established.
> logger.info("on_connecting: " + str(host_and_port))
>
> def on_disconnected(self):
> # Called by the STOMP connection when a TCP/IP connection to the
> STOMP server has been lost. No messages should be sent via the connection
> until it has been reestablished.
> logger.info("on_disconnected")
>
> def on_disconnecting(self):
> # Called before a DISCONNECT frame is sent.
> logger.info("on_disconnecting")
>
> def on_error(self,
> frame):
> # Called by the STOMP connection when an ERROR frame is received.
> # Parameters: frame (Frame) - the stomp frame
> logger.error("on_error: " + str(frame))
>
> def on_heartbeat(self):
> # Called on receipt of a heartbeat.
> logger.info("on_heartbeat")
>
> def on_heartbeat_timeout(self):
> # Called by the STOMP connection when a heartbeat message has not
> been received beyond the specified period.
> logger.info("on_heartbeat_timeout")
>
> def on_message(self,
> frame):
> # Called by the STOMP connection when a MESSAGE frame is received.
> # Parameters: frame (Frame) - the stomp frame
> # message is a stomp.utils.Frame with:
> # message.cmd == "MESSAGE"
> # message.headers == dict()
> # message.body == str
> logger.debug("on_message: " + str(frame))
>
> try:
> message_id = frame.headers["message-id"]
> subscription = int(frame.headers["subscription"])
>
> logger.info("Processing message with ID:" + str(message_id) + "
> and Subscription:" + str(subscription))
> # Simulate a long-running process -- much longer than TTL.
> for _ in range(0, 1200):
> # 100ms sleep intervals (should) keep the client from
> blocking.
> time.sleep(0.1)
> logger.info("Processed message with ID:" + str(message_id) + "
> and Subscription:" + str(subscription))
> self.conn.ack(message_id,
> subscription)
> logger.info("Message ACKed")
> except:
> logger.error(loggers.format_exc_info("Failed to process message:
> " + str(frame)))
> self.conn.nack(message_id,
> subscription)
> logger.info("Message NACKed")
>
> def on_receipt(self,
> frame):
> # Called by the STOMP connection when a RECEIPT frame is received,
> sent by the server if requested by the client using the ‘receipt’ header.
> # Parameters: frame (Frame) - the stomp frame
> logger.debug("on_receipt: " + str(frame))
>
> def on_receiver_loop_completed(self,
> frame):
> # Called when the connection receiver_loop has finished.
> logger.debug("on_receiver_loop_completed: " + str(frame))
>
> def on_send(self,
> frame):
> # Called by the STOMP connection when it is in the process of sending
> a message
> logger.debug("on_send: " + str(frame))
>
>
> if __name__ == "__main__":
> host="localhost"
> port=61613
> host_and_ports = [(host, port)]
> from_client_hb_millis = 30000
> from_server_hb_millis = 10000
> heartbeats = (from_client_hb_millis,
> from_server_hb_millis)
>
> logger.info("Listening to the AAA ActiveMQ queue on " +
> dumps(host_and_ports))
>
> # Connect and consume from the queue.
> conn = Connection12(host_and_ports=host_and_ports,
> heartbeats=heartbeats)
> listener = AAAHeartbeatListener(conn)
> conn.set_listener("",
> listener)
> conn.connect("admin",
> "admin",
> wait=True,
> headers = {"client-id": sys.argv[0]})
>
> # Stuff a few messages onto the queue.
> for _ in range(0,3):
> conn.send("AAA",
> dumps({"message": "How's yer mom en em?"}),
> "application/json",
> headers={"destination-type": "ANYCAST",
> "persistent": "true"})
>
> # We set the ActiveMQ prefetchSize to 1 so that only one message
> # is dispatched at a time.
> conn.subscribe("AAA",
> 42,
> ack="client-individual",
> headers={"subscription-type": "ANYCAST",
> "consumer-window-size": 0})
>
> # Keep the process alive until the connection terminates.
> try:
> while conn.is_connected():
> time.sleep(30)
> except:
> # Assuming a keyboard interrupt.
> pass
> finally:
> conn.disconnect()
>
> logger.info("I am dead.")
> -------
>
> If I run this as is I get:
>
> -------
> /usr/bin/python3.6 /pub/dev/flotsam/run_heartbeat_test.py
> 2022-06-24 09:23:06,187 __main__ INFO:
> Listening to the AAA ActiveMQ queue on [["localhost", 61613]]
>
> 2022-06-24 09:23:06,310 __main__ INFO:
> on_connecting: ('localhost', 61613)
>
> 2022-06-24 09:23:06,315 __main__ INFO:
> on_connected: {cmd=CONNECTED,headers=[{'version': '1.2', 'session':
> '672525f2', 'server': 'ActiveMQ-Artemis/2.22.0 ActiveMQ Artemis Messaging
> Engine', 'heart-beat': '10000,30000'}],body=}
>
> 2022-06-24 09:23:06,523 __main__ INFO:
> Processing message with ID:10205 and Subscription:42
>
> Heartbeat timeout: diff_receive=15.000172974017914, time=264585.48758635,
> lastrec=264570.487413376
> 2022-06-24 09:23:36,316 __main__ INFO:
> on_disconnected
>
> 2022-06-24 09:23:36,316 __main__ INFO:
> on_heartbeat_timeout
>
> 2022-06-24 09:23:36,335 __main__ INFO:
> I am dead.
>
>
> Process finished with exit code 0
> -------
>
> Note that the client does not process the server heartbeat while it is busy
> with the message, as if it's processing server heartbeats on the same thread
> as the one processing the long-running message.
>
> On the other hand, if I set from_server_hb_millis = 10000 I get the behavior
> I would expect -- the messages are processed, even though they take longer
> than TTL to complete -- and the listener stays alive until I kill it, as if
> it's sending heartbeats in another thread.
>
> Now for my purposes the latter scenario (not requesting server HBs) is all I
> need, but I just thought you might want to know about this with the hope that
> it's helpful to someone else down the line.
>
> Thank you!
>
> Regards,
>
> Rich Bergmann
>
> ________________________________
> From: Richard Bergmann <rbergm...@colsa.com.INVALID>
> Sent: Thursday, June 23, 2022 11:13 AM
> To: users@activemq.apache.org <users@activemq.apache.org>
> Subject: Re: [External] - Re: How to avoid AMQ229014 TTL "dump" for a
> "patient" Python 3 Stomp Listener
>
> CAUTION: This email originated from outside of the organization. Do not click
> links or open attachments unless you recognize the sender and know the
> content is safe.
>
>
> Thank you Justin. It's just now becoming apparent to me that the
> heartbeating mechanism is built into the client connection, i.e., I don't
> have to do anything on the on-heartbeat event. Doh!
>
> From what I've read I need to create a Connection12 to make all this go. So
> now I will proceed to try to get all this working.
>
> Thank you for your help and your patience.
>
> Regards,
>
> Rich Bergmann
> ________________________________
> From: Justin Bertram <jbert...@apache.org>
> Sent: Thursday, June 23, 2022 11:01 AM
> To: users@activemq.apache.org <users@activemq.apache.org>
> Subject: Re: [External] - Re: How to avoid AMQ229014 TTL "dump" for a
> "patient" Python 3 Stomp Listener
>
> CAUTION: This email originated from outside of the organization. Do not click
> links or open attachments unless you recognize the sender and know the
> content is safe.
>
>
> Heart-beating is something that your client implementation should provide
> for you automatically. You shouldn't need to send your own heart-beats
> manually. The client should also provide a way to configure the
> heart-beating on the initial connection so that it can send the proper
> header on the CONNECT frame. Does your Python STOMP client support
> heart-beating? The fact that you're not receiving any heart-beats indicates
> that either your client doesn't support it or you're not configuring it
> properly.
>
>
> Justin
>
>
> On Thu, Jun 23, 2022 at 9:22 AM Richard Bergmann
> <rbergm...@colsa.com.invalid> wrote:
>
> > I am set up to receive the heartbeat event on-heartbeat, but it is never
> > fired.
> >
> > Heartbeat makes sense, and I did a bit of digging on that, but the "STOMP
> > heart-beating and connection-ttl" section within
> > https://usg02.safelinks.protection.office365.us/?url=https%3A%2F%2Factivemq.apache.org%2Fcomponents%2Fartemis%2Fdocumentation%2Flatest%2Fstomp.html&data=05%7C01%7CRBERGMANN%40colsa.com%7C6a42b59f91384361fb6408da55f0f054%7C9821086b78824b43a5edb1e979bee31f%7C1%7C0%7C637916790583826130%7CUnknown%7CTWFpbGZsb3d8eyJWIjoiMC4wLjAwMDAiLCJQIjoiV2luMzIiLCJBTiI6Ik1haWwiLCJXVCI6Mn0%3D%7C3000%7C%7C%7C&sdata=2FWZF8PLzXfVQngRcyh3G7umv5eSo1ukHY%2FOq5SFoGM%3D&reserved=0
> > (which think I read maybe 50 times over the last two days?) basically sez
> >
> > (1) you need STOMP 1.1 or 1.2 (1.0 is the default)
> > (2) you need to send a heartbeat header
> > (3) you need to respond with a heartbeat
> >
> > But nowhere is does is specify WHAT a heartbeat message is!!!! Yes (and
> > just know), massive Googling presents me with no examples of HOW to do this.
> >
> > As I am using a basic STOMP Python connection (presumably 1.0), I am HAPPY
> > to send unsolicited heartbeats to the broker every, say, 30 seconds, but
> > I'm left wanting for just WHAT that heartbeat call is!!!!!
> >
> > Sorry, kinda frustrated here . . .
> > ________________________________
> > From: Justin Bertram <jbert...@apache.org>
> > Sent: Thursday, June 23, 2022 9:50 AM
> > To: users@activemq.apache.org <users@activemq.apache.org>
> > Subject: [External] - Re: How to avoid AMQ229014 TTL "dump" for a
> > "patient" Python 3 Stomp Listener
> >
> > CAUTION: This email originated from outside of the organization. Do not
> > click links or open attachments unless you recognize the sender and know
> > the content is safe.
> >
> >
> > > Is there a more acceptable way to alert the broker that the listener is
> > still alive and well but just being patient?
> >
> > Yes. STOMP already has this via heartbeating [1]. Does your Python client
> > support heartbeating? If so, are you configuring it?
> >
> >
> > Justin
> >
> > [1]
> > https://usg02.safelinks.protection.office365.us/?url=https%3A%2F%2Fstomp.github.io%2Fstomp-specification-1.2.html%23Heart-beating&data=05%7C01%7CRBERGMANN%40colsa.com%7C6a42b59f91384361fb6408da55f0f054%7C9821086b78824b43a5edb1e979bee31f%7C1%7C0%7C637916790583826130%7CUnknown%7CTWFpbGZsb3d8eyJWIjoiMC4wLjAwMDAiLCJQIjoiV2luMzIiLCJBTiI6Ik1haWwiLCJXVCI6Mn0%3D%7C3000%7C%7C%7C&sdata=G2esBzHWMNhtPuQdLshC1FOzLoNGaGS8jXlL%2Bi5Dwl4%3D&reserved=0
> >
> > On Thu, Jun 23, 2022 at 7:56 AM Richard Bergmann
> > <rbergm...@colsa.com.invalid> wrote:
> >
> > > I migrated my application from Classic to Artemis, and now my listeners
> > > are being summarily disconnected by the broker on a TTL timeout.
> > >
> > > My use case is that I have listeners (subscribers) that are launched as a
> > > service and patiently wait (could be for days!!!!) for messages to arrive
> > > on a work queue. Furthermore, the work to be done can take perhaps up to
> > > an hour . . . well past the default TTL timeout of 60 seconds. This was
> > > not an issue in Classic, but certainly is in Artemis.
> > >
> > > My solution was to bump the STOMP acceptor timeout to something
> > > stratospheric (e.g., 10,000,000,000), but this just feels . . . icky!
> > >
> > > I've seen a number of threads recommend doing a "connect and subscribe"
> > > loop in the on-disconnect event, but I have a bad feeling about this if
> > the
> > > listener is working on a message that is either NACKed on the disconnect,
> > > or the ACK fails because the work was completed between the disconnect
> > and
> > > the reconnect. In other words, this also feels icky!
> > >
> > > Is there a more acceptable way to alert the broker that the listener is
> > > still alive and well but just being patient? For example, is there a
> > NOOP
> > > call that can be made to the broker every, say, 30 seconds, to say "I'm
> > > still here! Please don't kill me!!!"? Since the listener may be in the
> > > throes of a long-running process, I can see having the listener spawn a
> > > thread that sends the "I'm alive!" message every 30 seconds, then,
> > > terminate the thread on a successful shutdown. That way the broker can
> > > kill legitimately "dead" listeners by keeping the reasonable 60 second
> > > timeout, but not assassinate my legitimate patient listeners.
> > >
> > > Thank you! :)
> > > ________________________________
> > > The information contained in this e-mail and any attachments from COLSA
> > > Corporation may contain company sensitive and/or proprietary information,
> > > and is intended only for the named recipient to whom it was originally
> > > addressed. If you are not the intended recipient, any disclosure,
> > > distribution, or copying of this e-mail or its attachments is strictly
> > > prohibited. If you have received this e-mail in error, please notify the
> > > sender immediately by return e-mail and permanently delete the e-mail and
> > > any attachments.
> > >
> > >
> > > COLSA Proprietary
> > >
> > ________________________________
> > The information contained in this e-mail and any attachments from COLSA
> > Corporation may contain company sensitive and/or proprietary information,
> > and is intended only for the named recipient to whom it was originally
> > addressed. If you are not the intended recipient, any disclosure,
> > distribution, or copying of this e-mail or its attachments is strictly
> > prohibited. If you have received this e-mail in error, please notify the
> > sender immediately by return e-mail and permanently delete the e-mail and
> > any attachments.
> >
> >
> > COLSA Proprietary
> >
> ________________________________
> The information contained in this e-mail and any attachments from COLSA
> Corporation may contain company sensitive and/or proprietary information, and
> is intended only for the named recipient to whom it was originally addressed.
> If you are not the intended recipient, any disclosure, distribution, or
> copying of this e-mail or its attachments is strictly prohibited. If you have
> received this e-mail in error, please notify the sender immediately by return
> e-mail and permanently delete the e-mail and any attachments.
>
>
> COLSA Proprietary
>
> COLSA Proprietary
> ________________________________
> The information contained in this e-mail and any attachments from COLSA
> Corporation may contain company sensitive and/or proprietary information, and
> is intended only for the named recipient to whom it was originally addressed.
> If you are not the intended recipient, any disclosure, distribution, or
> copying of this e-mail or its attachments is strictly prohibited. If you have
> received this e-mail in error, please notify the sender immediately by return
> e-mail and permanently delete the e-mail and any attachments.
>
>
> COLSA Proprietary
________________________________
The information contained in this e-mail and any attachments from COLSA
Corporation may contain company sensitive and/or proprietary information, and
is intended only for the named recipient to whom it was originally addressed.
If you are not the intended recipient, any disclosure, distribution, or copying
of this e-mail or its attachments is strictly prohibited. If you have received
this e-mail in error, please notify the sender immediately by return e-mail and
permanently delete the e-mail and any attachments.
COLSA Proprietary
