Mule 事务范围回滚似乎失败

Question

编辑：骡子 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>

编辑我尝试在第一个 INSERT 中添加一个 BEGIN_OR_JOIN 事务元素，在第二个中添加一个 ALWAYS_JOIN 事务元素，但是代码在到达第二个时抛出异常，即没有事务可以加入.

Answer 1

分别使用ALWAYS_BEGIN 和ALWAYS_JOIN 是要走的路。

但是，如果是两个不同的 DB，则需要使用 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>