Hi All, We had a discussion regarding the above implementation and identified an issue where the client side buffer would still contain messages even though we block messages from the MessageFlusher of the server using hasRoom flag. The messages on the client side buffer which can be received through the JMS message listener[1] or by using "receive"[2] method from the MessageConsumer.
Due to the above reason, the implementation for redelivery delay will be done on the client side(i.e client side andes jar). When a client receives a message, the message is placed in a queue(This is a LinkedBlockingQueue right now) within andes client jar. This queue will be used when a "receive"[2] method from JMS is called. The new proposed implementation is to use a DelayedQueue[3] for the above mentioned queue and only delay the redelivered messages. The delay amount can be set via a system property. When a message is acknowledged by a client that uses client acknowledgement, all the messages that were received and unacknowledged gets acknowledged. This is the expected behaviour as per in the specification[4]. But when this occurs, any message that that were supposed to get rejected by the client will get acked due to a subsequent valid message, given that the reject body has not been sent yet to the server. The correct way of handling this is by not accepting any new messages until the current message is fully processed. But in JMS we cant mark a message as rejected. Following is a sequence diagram for the mentioned scenario : [image: Screen Shot 2016-07-21 at 11.50.54 AM.png] The above issue can be fixed by introducing a new acknowledge mode which uses per message acknowledgment. Note that this is not a JMS acknowledgement value. Following is a sequence diagram with the fix : [image: Screen Shot 2016-07-21 at 12.00.00 PM.png] Notes : 1. Client can keep accept new messages regardless on whether the earlier message(s) was/were a reject/valid. 2. Rejected messages will be unordered when compared with valid messages. I.E Let's assume that the messages received to the andes client in the following order : ----> M1, M2, M3, M4, M5, M6 If M2 and M5 are invalid messages(to be rejected). The JMS client will receive the messages in the following order. ----> M1, M2, M3, M4, M5, M6, M2, M5, M2, M5..... M2 and M5 will be received repeatedly until they go to DLC. 3. The redelivery delay needs to be set by the client side using a system property. [1] - https://docs.oracle.com/javaee/7/api/javax/jms/MessageListener.html [2] - https://docs.oracle.com/javaee/7/api/javax/jms/MessageConsumer.html#receive-- [3] - https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/DelayQueue.html [4] - https://docs.oracle.com/cd/E19717-01/819-7757/aeqeg/index.html Regards, Hemika Hemika Kodikara Software Engineer WSO2 Inc. lean . enterprise . middleware http://wso2.com Mobile : +94777688882 On Wed, Jul 6, 2016 at 8:20 AM, Malaka Silva <[email protected]> wrote: > Hi Hemika, > > Do we maintain retry count and retry delay properties for this? > >> > On Tue, Jul 5, 2016 at 11:03 AM, Hemika Kodikara <[email protected]> wrote: > >> Hi All, >> >> Currently when a message is being rejected by a subscriber client, the >> client is able to delay returning the message to the server. This is done >> by setting the "AndesAckWaitTimeOut" system property. >> >> But the message cannot be delayed from the server side. I.E when a >> message rejected is received, the message is requeued straight away to the >> subscriber client. >> >> A delay should be able to added from the server side allowing the clients >> not being pushed with rejected messages. >> >> Solution is to have a configurable value in the broker.xml which applies >> a delay for each rejected message before being requeued. >> >> Each subscription(LocalSubscription) will have a delayed queue[1] which >> will store the rejected messages with the configuration value in >> broker.xml. When a rejected message is received, accepting new messages >> will be stopped from getting delivered to the client by using the >> LocalSubscription#hasRoomToAcceptMessages() method. New messages will only >> be accepted only when these rejected messages are accepted by the client or >> sent to the DLC. >> > Message can be rejected due to a business logic and can only be related > to a particular message. We should not stop processing other messages or > have this behavior configurable. > >> There will be a separate scheduled task which consists of subscriptions >> that has rejected messages. The scheduler will go through each of those >> subscriptions and deliver the messages to the outbound disruptor. >> >> >> >> >> See redmine for more details[2] >> >> [1] - >> https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/DelayQueue.html >> [2] - https://redmine.wso2.com/issues/5248 >> >> Regards, >> Hemika >> >> >> Hemika Kodikara >> Software Engineer >> WSO2 Inc. >> lean . enterprise . middleware >> http://wso2.com >> >> Mobile : +94777688882 >> >> _______________________________________________ >> Architecture mailing list >> [email protected] >> https://mail.wso2.org/cgi-bin/mailman/listinfo/architecture >> >> > > > -- > > Best Regards, > > Malaka Silva > Senior Technical Lead > M: +94 777 219 791 > Tel : 94 11 214 5345 > Fax :94 11 2145300 > Skype : malaka.sampath.silva > LinkedIn : http://www.linkedin.com/pub/malaka-silva/6/33/77 > Blog : http://mrmalakasilva.blogspot.com/ > > WSO2, Inc. > lean . enterprise . middleware > http://www.wso2.com/ > http://www.wso2.com/about/team/malaka-silva/ > <http://wso2.com/about/team/malaka-silva/> > https://store.wso2.com/store/ > > Save a tree -Conserve nature & Save the world for your future. Print this > email only if it is absolutely necessary. > > _______________________________________________ > Architecture mailing list > [email protected] > https://mail.wso2.org/cgi-bin/mailman/listinfo/architecture > >
_______________________________________________ Architecture mailing list [email protected] https://mail.wso2.org/cgi-bin/mailman/listinfo/architecture
