mule：バッチ処理のトランザクション

Question

トピックにサブスクライブするJMSインバウンドエンドポイントがあります。メッセージトランスフォーマーがあると、ペイロードをレコードのリストに分割し、バッチコミットを使用してデータベースに挿入します。データベースに挿入中にエラーが発生した場合は、ペイロード全体をJMSにロールバックする必要があります。どのようにトランザクションを使用してこれを達成するには？

<batch:job name="lockboxBatch" max-failed-records="-1"> 
     <batch:input> 
      <jms:inbound-endpoint topic="lockbox" connector-ref="Active_MQ1" doc:name="JMS"> 
      </jms:inbound-endpoint> 
      <custom-transformer class="transformers.PaymentsTransformer" doc:name="Java"/> 
      <logger level="INFO" doc:name="Logger"/> 
     </batch:input> 
     <batch:process-records> 
      <batch:step name="Batch_Step"> 
       <expression-component doc:name="Expression"><![CDATA[payload[2].batchAmount='hghghfghfhgf']]></expression-component> 
       <batch:commit size="4" doc:name="Batch Commit"> 
        <db:insert config-ref="Oracle_Configuration" doc:name="Database" bulkMode="true" > 
         <db:parameterized-query><![CDATA[INSERT INTO TIB_INT_AR_PAYMENT_IFACE (TRANSMISSION_REQUEST_ID,DESTINATION_ACCOUNT,ORIGINATION,TRANSMISSION_RECORD_COUNT,TRANSMISSION_AMOUNT,LOCKBOX_NUMBER,LOCKBOX_BATCH_COUNT,LOCKBOX_RECORD_COUNT,LOCKBOX_AMOUNT,BATCH_NAME,BATCH_AMOUNT,BATCH_RECORD_COUNT,ITEM_NUMBER,CURRENCY_CODE,REMITTANCE_AMOUNT,TRANSIT_ROUTING_NUMBER,ACCOUNT,CHECK_NUMBER,CUSTOMER_NUMBER,OVERFLOW_INDICATOR,OVERFLOW_SEQUENCE,INVOICE1,AMOUNT_APPLIED1) VALUES (#[payload.?transmissiosnRequestID],#[payload.?destinastionAccount],#[payload.?origination],#[payload.?transmissionSrecordCount],#[payload.?transmisssionAmount],#[payload.lockboxNumber],#[payload.lockboxBatchCount],#[payload.lockboxRecordCount],#[payload.lockboxAmount],#[payload.batchName],#[payload.batchAmount],#[payload.batchRecordCount],#[payload.itemNumber],#[payload.currencyCode],#[payload.remittanceAmount],#[payload.transitRoutingNumber],#[payload.account],#[payload.checkNumber],#[payload.customerNumber],#[payload.overflowIndicator],#[payload.overflowSequence],#[payload.invoice1],#[payload.amountApplied1])]]></db:parameterized-query> 
        </db:insert> 
       </batch:commit> 
      </batch:step> 
      <batch:step name="Batch_Step1" accept-policy="ONLY_FAILURES"> 
      <set-payload value="#[getStepExceptions()]" doc:name="Set Payload"/> 
       <foreach collection="#[payload.values()]" doc:name="For Each"> 
        <jms:outbound-endpoint queue="Invalid_Transmission" connector-ref="Active_MQ" doc:name="JMS"/> 
       </foreach> 
      </batch:step> 


     </batch:process-records> 
     <batch:on-complete> 
      <logger message="Completed the insert" level="INFO" doc:name="Logger"/> 
     </batch:on-complete> 
    </batch:job> 

Answer 1

2つのことを試すことができます。

1.Make max-failed-records = "0"これは、バッチステップでエラーが発生した場合にロールバックします。

2.トランザクションスコープ内のDBコネクタをエミュレートし、必要に応じて例外シナリオを処理します。

<transactional action="ALWAYS_BEGIN" doc:name="Transactional"> 
    <db:insert...>........</db:insert> 
</transactional> 

下記の更新コードを考慮して、要件に合わせて変更することができます。

<batch:job name="lockboxBatch" max-failed-records="0"> 
    <batch:input> 
     <jms:inbound-endpoint topic="lockbox" connector-ref="Active_MQ1" doc:name="JMS"> 
     </jms:inbound-endpoint> 
     <custom-transformer class="transformers.PaymentsTransformer" doc:name="Java"/> 
     <logger level="INFO" doc:name="Logger"/> 
    </batch:input> 
    <batch:process-records> 
     <batch:step name="Batch_Step"> 
      <expression-component doc:name="Expression"><![CDATA[payload[2].batchAmount='hghghfghfhgf']]></expression-component> 
      <batch:commit size="4" doc:name="Batch Commit"> 
        <transactional action="ALWAYS_BEGIN" doc:name="Transactional"> 
         <db:insert config-ref="Oracle_Configuration" bulkMode="true" doc:name="Database"> 
          <db:parameterized-query><![CDATA[INSERT INTO TIB_INT_AR_PAYMENT_IFACE (TRANSMISSION_REQUEST_ID,DESTINATION_ACCOUNT,ORIGINATION,TRANSMISSION_RECORD_COUNT,TRANSMISSION_AMOUNT,LOCKBOX_NUMBER,LOCKBOX_BATCH_COUNT,LOCKBOX_RECORD_COUNT,LOCKBOX_AMOUNT,BATCH_NAME,BATCH_AMOUNT,BATCH_RECORD_COUNT,ITEM_NUMBER,CURRENCY_CODE,REMITTANCE_AMOUNT,TRANSIT_ROUTING_NUMBER,ACCOUNT,CHECK_NUMBER,CUSTOMER_NUMBER,OVERFLOW_INDICATOR,OVERFLOW_SEQUENCE,INVOICE1,AMOUNT_APPLIED1) VALUES (#[payload.?transmissiosnRequestID],#[payload.?destinastionAccount],#[payload.?origination],#[payload.?transmissionSrecordCount],#[payload.?transmisssionAmount],#[payload.lockboxNumber],#[payload.lockboxBatchCount],#[payload.lockboxRecordCount],#[payload.lockboxAmount],#[payload.batchName],#[payload.batchAmount],#[payload.batchRecordCount],#[payload.itemNumber],#[payload.currencyCode],#[payload.remittanceAmount],#[payload.transitRoutingNumber],#[payload.account],#[payload.checkNumber],#[payload.customerNumber],#[payload.overflowIndicator],#[payload.overflowSequence],#[payload.invoice1],#[payload.amountApplied1])]]></db:parameterized-query> 
         </db:insert> 
        </transactional> 

      </batch:commit> 
     </batch:step> 
     <batch:step name="Batch_Step1" accept-policy="ONLY_FAILURES"> 
     <set-payload value="#[getStepExceptions()]" doc:name="Set Payload"/> 
      <foreach collection="#[payload.values()]" doc:name="For Each"> 
       <jms:outbound-endpoint queue="Invalid_Transmission" connector-ref="Active_MQ" doc:name="JMS"/> 
      </foreach> 
     </batch:step> 


    </batch:process-records> 
    <batch:on-complete> 
     <logger message="Completed the insert" level="INFO" doc:name="Logger"/> 
    </batch:on-complete> 
</batch:job>