骡子事务回滚范围出现失败

Question

编辑：骡子3.4.1

我们有一个骡子流是交替从一个数据库插入到另一个，都包裹事务的范围内读取。在这种特殊情况下，后面的插入之一会失败，我们期望所有内容都回滚。

当我们查看日志时，第二次插入（即下例中的BulkInsertInstanceToCache）会出现异常（例如重复的PRIMARY KEY）。当我们查看数据库时，会看到第一个插入的数据（下例中的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>