[Java/EJB3.0] Message Driven Bean; probleem met transacties

[Michali]

Het probleem:

Ik ben aan het spelen met EJB 3.0 en ik ben op een probleem gestuit waar ik echt niet uit kom. Het gaat om het afhandelen van berichten door een Message Driven Bean. Het is de bedoeling dat deze class, bij het ontvangen van een berichtje, een bedrag van één account naar de andere overschrijft. 2 bestaande Account objecten worden dus opgehaald, waarvan dan het balans (het bedrag) wordt aangepast.

Het is van groot belang dat dit middels een transactie gebeurt, anders is het mogelijk dat 2 berichten ervoor zorgen dat een Account 2 keer wordt uitgelezen en opgeslagen, en dat zo het balans niet meer klopt. Juist dit is het probleem, want ik krijg het met geen mogelijkheid in een transactie.

Als ik bijvoorbeeld 3 berichten verzend, dan worden ze alle 3 geaccepteerd door verschillende instanties van de Message Driven Bean class (concurrent dus). Dan worden de Account objecten opgehaald en gewijzigd, waarna de transacties één voor één aflopen doordat de functie beëindigd. Dit betekent dus meestal dat alleen het laatst gestuurde bericht effect heeft op het balans, omdat deze simpelweg de wijzigingen van de andere 2 overschrijft.


Wat heb ik geprobeerd:

Ik weet dat Message Driven Beans standaard door de container transacties krijgen toegewezen die standaard van het type REQUIRED zijn. Dit lijkt echter de problemen niet te voorkomen. Als ik zelf controle neem over de transacties (middels een EntityTransaction), dan gebeurt precies hetzelfde.

Ik heb ook verschillende manieren van het verzenden van een berichtje geprobeerd (zoals het voor ieder bericht opnieuw opzetten van een connection), maar dit had (zoals verwacht) geen effect.

Ik heb heel veel gezocht, maar ik word er echt niet wijs uit, helaas.


Waar ben ik naar op zoek?

Als ik er voor kan zorgen dat er maar echt 1 bericht tegelijk wordt verwerkt, dan ben ik al redelijk van het probleem af, maar dat kan later weer voor problemen zorgen. Wat ik eigenlijk dus wil, is dat de ingelezen objecten als het ware gelocked worden, totdat de wijziging is gedaan en het resultaat is opgeslagen in de database (de transactie is gecommit dus). Bij mijn weten is dit redelijk standaard bij een transactie, dus waarom dit dan nu niet werkt is me echt een raadsel. Hoe zorg ik er dus voor dat de 3 berichten zonder problemen verwerkt worden?


Code voorbeelden:

Sender:

Java:

InitialContext jndiContext = new InitialContext();
QueueConnectionFactory connectionFactory = (QueueConnectionFactory)jndiContext.lookup("java:/JmsXA");
Queue queue = (Queue)jndiContext.lookup("queue/AccountTransactionQueue");

QueueConnection connection = connectionFactory.createQueueConnection();
QueueSession session = connection.createQueueSession(true, 0);
QueueSender sender = session.createSender(queue);

for ( PendingTransaction pendingTransaction : pendingTransactions )
{
    MapMessage mapMessage = session.createMapMessage();
    mapMessage.setInt("fromAccount", fromAccountNumber);
    mapMessage.setInt("toAccount", pendingTransaction.toAccountNumber);
    mapMessage.setDouble("amount", pendingTransaction.amount);
    sender.send(mapMessage);
}

sender.close();
session.close();
connection.close();

Message Driven Bean:

Java:

@MessageDriven(
    activationConfig={
        @ActivationConfigProperty(
            propertyName="destinationType",
            propertyValue="javax.jms.Queue"
        ),
        @ActivationConfigProperty(
            propertyName="destination",
            propertyValue="queue/AccountTransactionQueue"
        )
    }
)
public class AccountTransactionManager implements javax.jms.MessageListener
{
    @PersistenceContext(unitName="AccountsDB")
    private EntityManager entityManager;

    public void onMessage(Message message)
    {
        MapMessage mapMessage = (MapMessage)message;
        int fromAccountNumber = mapMessage.getInt("fromAccount");
        int toAccountNumber = mapMessage.getInt("toAccount");
        double amount = mapMessage.getDouble("amount");

        Account fromAccount = (Account)entityManager.find(Account.class, fromAccountNumber);
        Account toAccount = (Account)entityManager.find(Account.class, toAccountNumber);

        if ( fromAccount == null || toAccount == null )
        {
            throw new Exception();
        }

        fromAccount.setBalance(fromAccount.getBalance() - amount);
        toAccount.setBalance(toAccount.getBalance() + amount);
    }
}

Disclaimer: De 100% correctheid of veiligheid (behalve die transacties dan) van deze code is even geen issue, wat ik graag wil is dat ik het probleem snap, dat staat voorop. Dit is even een stuk versimpeld dus.

[voodooless]

Kun je i.p.v X messages niet gewoon 1 ,message sturen die de info bevat voor al die losse Accountmutaties. Dan heb je een message die dat voor je doet ipv X.

[Michali]

Dat kan idd., maar dat lost het probleem toch niet op? Dit probleem zou nog steeds kunnen voorkomen als er 2 berichten snel na elkaar worden verstuurd, of als er bijvoorbeeld tegelijkertijd geld van en naar een rekening wordt geschreven.


Ik pas nu overigens locking toe, met EntityManager.lock(object, LockModeType.WRITE), wat wel voorkomt dat de gegevens inconsistent worden. Maar dit zorgt er wel voor dat bepaalde berichten gewoon niet worden verwerkt, maar eindigen met een exception. Dat kan ook niet de bedoeling zijn.

[voodooless]

Je kunt je message synchronized maken rond een statisch object, zo kan er maar een van een dergelijke message tegelijk uitgevoerd worden:

Java:

...
public class AccountTransactionManager implements javax.jms.MessageListener
{
   static Object synObject = new Object();

   public void onMessage(Message message)
   { 
      synchronized(synObject){
         // do stuff...
      }
   }

Geen idee of het echt gaat werken zo, maar het is het proberen waard!

[ Voor 9% gewijzigd door voodooless op 18-04-2007 10:05 ]

[kasper_vk]

Michali schreef op woensdag 18 april 2007 @ 08:59:
... Maar dit zorgt er wel voor dat bepaalde berichten gewoon niet worden verwerkt, maar eindigen met een exception. Dat kan ook niet de bedoeling zijn.

Die berichten worden dan toch weer opnieuw bezorgd? Zo niet, dan is er kennelijk geen sprake van transactionele verwerking begrijp ik uit de documentatie van de Message en en MessageHandler interfaces.
Daaruit begrijp ik namelijk dat een bericht dat wel is bezorgd, maar niet is bevestigd (acknowledge() methode) later opnieuw wordt bezorgd. Verder staat er dat binnen een transactionele context het aanroepen van acknowledge() wordt genegeerd, aangezien de transactie-manager zelf acknowledge() aanroept wanneer de transactie commit.

Kun je hier iets mee?
Ik vraag me dus af of die onMessage() wel echt transactioneel verwerkt wordt, zoals je zou willen.

[JKVA]

@voodooless:
Synchronized code in een EJB is niet toegestaan. Net als het knutselen met Threads.

@Michali:
Met MDB's weet je niet altijd waar je aan toe bent. Dat is voor een deel ook de kracht, want het koppelt je componenten helemaal los van elkaar.

Uit de specificatie:

Only the NOT_SUPPORTED and REQUIRED transaction attributes may be used for message-driven bean message listener methods. The use of the other transaction attributes is not meaningful for message-driven bean message listener methods because there is no pre-existing client transaction context (REQUIRES_NEW, SUPPORTS) and no client to handle exceptions (MANDATORY, NEVER).

NOT_SUPPORTED
The container invokes a message-driven bean message listener method whose transaction attribute is set to NOT_SUPPORTED with an unspecified transaction context. If the message listener method invokes other enterprise beans, the container passes no transaction context with the invocation.

REQUIRED
The container must invoke a message-driven bean message listener method whose transaction attribute is set to REQUIRED with a valid transaction context. The resource managers accessed by the message listener method within the transaction are enlisted with the transaction. If the message listener method invokes other enterprise beans, the container passes the transaction context with the invocation. The container attempts to commit the transaction when the message listener method has completed. Messaging systems may differ in quality of service with regard to reliability and transactionality of the dequeuing of messages.
The requirement for JMS are as follows:

A transaction must be started before the dequeuing of the JMS message and, hence, before the invocation of the message-driven bean’s onMessage method. The resource manager associated with the arriving message is enlisted with the transaction as well as all the resource managers accessed by the onMessage method within the transaction. If the onMessage method invokes other enterprise beans, the container passes the transaction context with the invocation. The transaction is committed when the onMessage method has completed. If the onMessage method does not successfully complete or the transaction is rolled back, message redelivery semantics apply.

Message Driven Beans draaien nooit mee in de transactie die op de client start. Ze worden wel opnieuw gestuurd (tot ze op de dead letter queue komen) als er geen ack komt. Die ack wordt door de container (bij CMT) bij transactie commit verstuurd.

Kortom, je moet geen volgorde-afhankelijke zaken in een MDB stoppen...

[voodooless]

Tjonge.. Dat zijn nogal wat voorwaarden zeg... Voor ons inieder geval meer dan zat rede om van dergelijke technologien af te zien, kracht of niet...

[kasper_vk]

JKVA schreef op woensdag 18 april 2007 @ 16:28:
...
Kortom, je moet geen volgorde-afhankelijke zaken in een MDB stoppen...

Volgens mij trek je een verkeerde conclusie: in een MDB (in de onMessage methode) kun je gerust volgorde afhankelijke zaken regelen (de implementatie van onMessage blijft immers gewoon code).

Wanneer je zou stellen dat je geen volgorde-afhankelijke taken moet uitvoeren via meerdere messages (en dan hoopt op een volgordelijke afhandeling van verschillende messages), geef ik je groot gelijk .

Naar mijn idee is het probleem van de TS hiermee dus verklaard noch opgelost; al zou het fijn zijn al hij daar zelf (weer) eens iets over riep

[Michali]

Als er geen volgorde afhankelijke zaken in kunnen, waarom worden dan transacties ondersteund? Die zijn toch juist bedoeld voor dit soort zaken?

Zoals gezegd lock ik de entities nu met EntityManager.lock. Dit zorgt ervoor dat als andere instanties in hun eigen thread ook een lock aanvragen, er een exception wordt gegooid. In de logs zie ik dan dat hij idd. nog 1 keer verstuurd wordt, en dat hij daarna ongeldig verklaard wordt.

Ik had overigens verwacht dat het volgende bericht pas verwerkt zou worden als de acknowlegde terug gestuurd wordt. Blijkbaar wordt het volgende bericht alweer aan een andere instantie aangeboden voordat dit is gebeurd, anders zouden de berichten niet tegelijkertijd verwerkt worden. Als ik er over nadenk is het ook wel enigzins logisch, want dit scheelt je natuurlijk wel in performance.

Wat ik niet logisch vindt, is dat een andere transactie met entities kan werken, die al in een andere transactie gebruikt worden. Dat vraagt toch om problemen? Met locken zou je dit kunnen voorkomen, maar dan zou ik het veel logischer vinden dat de executie van code op dat punt blokeert totdat er wel een lock verkregen kan worden. In plaats daarvan wordt er gewoon een exception gegooid, wat resulteert in het niet kunnen verwerken van een bericht. Lekker is dat.

[ari3]

Michali schreef op donderdag 19 april 2007 @ 10:01:
Als er geen volgorde afhankelijke zaken in kunnen, waarom worden dan transacties ondersteund? Die zijn toch juist bedoeld voor dit soort zaken?

Jawel, maar je moet natuurlijk wel kunnen aangegeven waar een transactie begint en waar hij eindigt. Dat kan in jouw use-case alleen maar als je de transactiecontext bewaart en bij binnenkomst van een twee, derde message weer kan koppelen aan de oorspronkelijke transactie. Dit is een niet-gangbaar ontwerp omdat transacties onder een timeout-beleid zitten. Als je de transactiecontext gaat bewaren in afwachting van een volgend bericht dan heb je kans dat de transactie (1) timeout of (2) andere processen blokkeert wegens locking.

Ik had overigens verwacht dat het volgende bericht pas verwerkt zou worden als de acknowlegde terug gestuurd wordt. Blijkbaar wordt het volgende bericht alweer aan een andere instantie aangeboden voordat dit is gebeurd, anders zouden de berichten niet tegelijkertijd verwerkt worden. Als ik er over nadenk is het ook wel enigzins logisch, want dit scheelt je natuurlijk wel in performance.

Dat is inderdaad zoals het werkt. Een alternatieve oplossing is dan het aantal queue-workers terug te brengen naar 1 zodat je zeker weet dat berichten niet gelijktijdig worden verwerkt.

Wat ik niet logisch vindt, is dat een andere transactie met entities kan werken, die al in een andere transactie gebruikt worden. Dat vraagt toch om problemen? Met locken zou je dit kunnen voorkomen, maar dan zou ik het veel logischer vinden dat de executie van code op dat punt blokeert totdat er wel een lock verkregen kan worden. In plaats daarvan wordt er gewoon een exception gegooid, wat resulteert in het niet kunnen verwerken van een bericht. Lekker is dat.

Met die exceptie heb je toch een signaal dat er een lock is? Je kunt dan zelf "wacht-tot-lock-opgeheven-is" implementeren. Ik verwacht niet dat een applicatieserver gaat pollen totdat een lock is opgeheven. Dat is niet performant en is daarmee geen gangbaar ontwerpprincipe.

[Michali]

ari3 schreef op donderdag 19 april 2007 @ 11:01:
[...]
Jawel, maar je moet natuurlijk wel kunnen aangegeven waar een transactie begint en waar hij eindigt. Dat kan in jouw use-case alleen maar als je de transactiecontext bewaart en bij binnenkomst van een twee, derde message weer kan koppelen aan de oorspronkelijke transactie. Dit is een niet-gangbaar ontwerp omdat transacties onder een timeout-beleid zitten. Als je de transactiecontext gaat bewaren in afwachting van een volgend bericht dan heb je kans dat de transactie (1) timeout of (2) andere processen blokkeert wegens locking.

Even voor de duidelijkheid, waar ik dus niet naar op zoek ben, is het verwerken van meerdere berichten in 1 transactie. Ieder bericht wordt behandeld in zijn eigen transactie.

Dat is inderdaad zoals het werkt. Een alternatieve oplossing is dan het aantal queue-workers terug te brengen naar 1 zodat je zeker weet dat berichten niet gelijktijdig worden verwerkt.

Heb je misschien een voorbeeldje of een duwtje in de goede richting, voor hoe je dit configureert?

Met die exceptie heb je toch een signaal dat er een lock is? Je kunt dan zelf "wacht-tot-lock-opgeheven-is" implementeren. Ik verwacht niet dat een applicatieserver gaat pollen totdat een lock is opgeheven. Dat is niet performant en is daarmee geen gangbaar ontwerpprincipe.

Dat zou idd. kunnen. Dan zou je met een loop moeten werken, waarbij pas gebroken wordt als het bericht succesvol verwerkt is. Maar hoe netjes is dat? Is dat een gebruikelijke manier van werken, of probeer ik een MDB echt te gebruiken voor zaken waarvoor hij niet bedoeld is?

[ari3]

Michali schreef op vrijdag 20 april 2007 @ 08:39:
Heb je misschien een voorbeeldje of een duwtje in de goede richting, voor hoe je dit configureert?

Dat is specifiek voor iedere JMS-implementatie dacht ik. Verschilt dus per applicatieserver. In JBoss 4 kun je jbossmq-service.xml aanpassen zodat de ThreadPool mbean max. 1 thread bevat én een "wait" blockingmodus heeft zodat gewacht wordt met verwerken totdat er weer een thread beschikbaar is in de poel.

Dat zou idd. kunnen. Dan zou je met een loop moeten werken, waarbij pas gebroken wordt als het bericht succesvol verwerkt is. Maar hoe netjes is dat? Is dat een gebruikelijke manier van werken, of probeer ik een MDB echt te gebruiken voor zaken waarvoor hij niet bedoeld is?

Dat is inderdaad niet netjes. Zou je die weg kiezen dan zou je een javax.management.timer.TimerMBean implementatie kunnen toepassen. Zo'n mbean kan dan periodiek (iedere 100ms) polsen of de lock al verdwenen is.

[Verwijderd]

Ik denk eigenlijk stiekem dat je het simpelweg helemaal niet moet willen. Doordat je wilt locken 'moet' je overige berichten in de wacht zetten. Je loopt daarmee een aannemelijk risico dat je wachtrij overstroomt.

Ik zou dus serieus een andere aanpak overwegen. Bijvoorbeeld door in een tijdelijk opslag een transactie object op te bouwen of door data tijdelijk 'stale' te noemen oid. Maar hoe dan ook nooit andere request uitsluiten of blokkeren omdat je op een verzameling van berichten zit te wachten.

[JKVA]

kasper_vk schreef op donderdag 19 april 2007 @ 09:16:
[...]

Volgens mij trek je een verkeerde conclusie: in een MDB (in de onMessage methode) kun je gerust volgorde afhankelijke zaken regelen (de implementatie van onMessage blijft immers gewoon code).

Eens. Dat bedoelde ik eigenlijk ook, maar de notatie liet wat te wensen over.

Michali schreef op donderdag 19 april 2007 @ 10:01:
Als er geen volgorde afhankelijke zaken in kunnen, waarom worden dan transacties ondersteund? Die zijn toch juist bedoeld voor dit soort zaken?

Zoals gezegd lock ik de entities nu met EntityManager.lock. Dit zorgt ervoor dat als andere instanties in hun eigen thread ook een lock aanvragen, er een exception wordt gegooid. In de logs zie ik dan dat hij idd. nog 1 keer verstuurd wordt, en dat hij daarna ongeldig verklaard wordt.

Ik had overigens verwacht dat het volgende bericht pas verwerkt zou worden als de acknowlegde terug gestuurd wordt. Blijkbaar wordt het volgende bericht alweer aan een andere instantie aangeboden voordat dit is gebeurd, anders zouden de berichten niet tegelijkertijd verwerkt worden. Als ik er over nadenk is het ook wel enigzins logisch, want dit scheelt je natuurlijk wel in performance.

Wat ik niet logisch vindt, is dat een andere transactie met entities kan werken, die al in een andere transactie gebruikt worden. Dat vraagt toch om problemen? Met locken zou je dit kunnen voorkomen, maar dan zou ik het veel logischer vinden dat de executie van code op dat punt blokeert totdat er wel een lock verkregen kan worden. In plaats daarvan wordt er gewoon een exception gegooid, wat resulteert in het niet kunnen verwerken van een bericht. Lekker is dat.

In principe is het niet vreemd dat je in de ene transactie met een object (record) werkt en in een andere ook. Dat hangt maar net af van je isolation level. Ik weet niet zeker of dit hier bij jouw probleem speelt, maar het hoeft niet vreemd te zijn.