Mule транзакционной сфера Откат, как представляется, не в состоянии

Question

Edit: Mule 3.4.1

У нас есть поток Mule, который чередуется считывает из одной базы данных со вставками в другую, все обернутые в транзакционной области. В этом конкретном случае одна из более поздних вставок терпит неудачу, и мы ожидаем, что все откатится.

Когда мы смотрим в журналах, мы видим исключение (например, дублируем PRIMARY KEY) для второй вставки (то есть BulkInsertInstanceToCache в примере ниже). Когда мы смотрим в базу данных, мы видим данные из первой вставки (BulkInsertActivityToCache в примере ниже). Мы ожидали, что все данные исчезнут.

Мы неправильно настроили эту область для того, что мы хотим?

Вот пример кода, где я сократил его до двух вставок, чтобы показать тип выполняемой обработки.

<flow name="ProcessBulkUpdateCache" processingStrategy="synchronous" doc:name="ProcessBulkUpdateCache"> 
    <transactional action="ALWAYS_BEGIN" doc:name="Transactional"> 
     <jdbc-ee:outbound-endpoint exchange-pattern="request-response" 
       queryKey="GetActivitiesForCache" queryTimeout="-1" connector-ref="SumTotalDatabase"> 
      <jdbc-ee:transaction action="NONE" /> 
     </jdbc-ee:outbound-endpoint> 

     <jdbc-ee:outbound-endpoint exchange-pattern="request-response" 
       queryKey="BulkInsertActivityToCache" queryTimeout="-1" connector-ref="EAIServiceDatabase"> 
     </jdbc-ee:outbound-endpoint> 

     <jdbc-ee:outbound-endpoint exchange-pattern="request-response" 
       queryKey="GetInstancesForCache" queryTimeout="-1" connector-ref="SumTotalDatabase"> 
      <jdbc-ee:transaction action="NONE" /> 
     </jdbc-ee:outbound-endpoint>  

     <jdbc-ee:outbound-endpoint exchange-pattern="request-response" 
       queryKey="BulkInsertInstanceToCache" queryTimeout="-1" connector-ref="EAIServiceDatabase"> 
     </jdbc-ee:outbound-endpoint>   
    </transactional> 

    <catch-exception-strategy doc:name="Unexpected"> 
     ...etc. 
    </catch-exception-strategy>     
</flow> 

Редактировать Я попытался добавить элемент BEGIN_OR_JOIN транзакции в первом INSERT и элемент транзакции ALWAYS_JOIN во втором, но код бросает исключение, когда он достигнет второй, что не существует ни одной сделки с открытым присоединиться.

Answer 1

Использование ALWAYS_BEGIN и ALWAYS_JOIN соответственно - путь.

Но если это две разные БД, вам нужно использовать транзакции XA. Локальная транзакция не может регистрировать ресурсы из двух разных баз данных.

Answer 2

Как я знаю, первое исходящее должно действие быть всегда-начал, там после того, как каждый исходящий должен быть всегда, начал

<db:mysql-config name="MySQL_Configuration" host="localhost" port="3306" user="root" database="ib_trade" useXaTransactions="true" driverClassName="com.mysql.jdbc.Driver" doc:name="MySQL Configuration"/> 
    <flow name="transactonmanagerFlow"> 
     <http:listener config-ref="HTTP_Listener_Configuration" path="/ram" doc:name="HTTP"/> 
     <logger level="INFO" doc:name="Logger" message="%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%"/> 
     <set-payload value="#['hi']" doc:name="Set Payload"/> 
     <vm:outbound-endpoint exchange-pattern="request-response" path="temp" connector-ref="VM" doc:name="VM"> 
      <xa-transaction action="ALWAYS_JOIN" timeout="10000"/> 
     </vm:outbound-endpoint> 
     <logger level="INFO" doc:name="Logger" message="^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^"/> 
    </flow> 
    <flow name="transactonmanagerFlow1"> 
     <vm:inbound-endpoint exchange-pattern="request-response" path="temp" connector-ref="VM" doc:name="VM"> 
      <xa-transaction action="ALWAYS_BEGIN" timeout="100000"/> 
     </vm:inbound-endpoint> 
     <logger level="INFO" doc:name="Logger" message="**************************************************************************"/> 
     <db:insert config-ref="MySQL_Configuration" transactionalAction="ALWAYS_JOIN" doc:name="Database"> 
      <db:dynamic-query><![CDATA[INSERT INTO `ib_trade`.`swt_symbol`(`idswt_symbol`,`symbol_name`,`symbol_exchange`,`symbol_id`) VALUES ("5","1","1","1");]]></db:dynamic-query> 
     </db:insert> 
     <logger message="^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$" level="INFO" doc:name="Logger"/> 
     <db:insert config-ref="MySQL_Configuration" transactionalAction="ALWAYS_JOIN" doc:name="Copy_of_Database"> 
      <db:dynamic-query><![CDATA[INSERT INTO `ib_trade`.`swt_symbol`(`idswt_symbol`,`symbol_name`,`symbol_exchange`,`symbol_id`) VALUES ("5","temp","1","1");]]></db:dynamic-query> 
     </db:insert>