騾子事務回滾範圍出現失敗

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>