Mule ESB 3.3 异常策略和回滚莫名其妙的行为

Question

设置：
表：CREATE TABLE test (id int PRIMARY KEY NOT NULL, name varchar(25));

基本测试流程：

前提条件：数据库已有id=2的记录（强制SqlException）

案例 1：

按原样调用基本流程时，由于主键冲突，第二次插入失败，第一次插入回滚。似乎是一种合理的默认行为。

案例 2：

通过添加回滚异常策略来修改基本流程，并在异常策略中添加一个记录器，以便在调用它时打印一些内容。
调用流时，由于主键冲突，第二次插入失败，第一次插入被回滚但永远不会调用 RollbackExceptionStrategy！
几乎没有人会期待。

所以这里的问题是：为什么不调用回滚异常策略以及我需要做什么才能调用它？

案例 3：

通过添加捕获异常策略来修改基本流程，并在异常策略中添加一个记录器，以便在调用它时打印一些内容。
调用流时，由于主键违规，第二次插入失败。这次调用了异常策略，但现在事务没有回滚。

这里的问题是：为什么事务没有被回滚以及如何在不同于回滚策略的异常策略中强制回滚？

任何帮助将不胜感激

---

Edit1：这是流程的完整 xml（仅基本情况）：

<mule>
    <spring:beans>
        <spring:bean id="dataSource" name="dataSource" class="org.enhydra.jdbc.standard.StandardDataSource" destroy-method="shutdown">
            <spring:property name="driverName" value="org.h2.Driver" />
            <spring:property name="url" value="jdbc:h2:tcp://localhost/~/mule" />
            <spring:property name="user" value="sa" />
            <spring:property name="password">
                <spring:value></spring:value>
            </spring:property>
        </spring:bean>

        <spring:bean id="transactionFactory" name="transactionFactory" class="org.mule.transport.jdbc.JdbcTransactionFactory" />

    </spring:beans>

    <jdbc:connector name="dbConnector" dataSource-ref="dataSource" validateConnections="true" queryTimeout="-1" pollingFrequency="0" doc:name="Database" />

    <flow name="TriggerFlow" doc:name="TriggerFlow">
        <http:inbound-endpoint exchange-pattern="request-response" host="localhost" port="8081" doc:name="HTTP" />
        <vm:outbound-endpoint exchange-pattern="request-response" path="txFlow" doc:name="VM" />
    </flow>

    <flow name="TxFlow" doc:name="case1Flow">
        <vm:inbound-endpoint exchange-pattern="request-response" path="txFlow" doc:name="case1">
            <custom-transaction factory-ref="transactionFactory" action="ALWAYS_BEGIN" timeout="10" />
        </vm:inbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert" queryTimeout="-1" connector-ref="dbConnector" doc:name="insert into test values (1, 'Test 1')">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert" value="insert into test values (1, 'Test 1')" />
        </jdbc:outbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert2" queryTimeout="-1" connector-ref="dbConnector" doc:name="insert into test values (2, 'Test 2')">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert2" value="insert into test values (2, 'Test 2')" />
        </jdbc:outbound-endpoint>
    </flow>

</mule>

---

Edit2：事实证明，案例 2 孤立地工作，但它不适用于更复杂的流程，例如：

<mule>
    <spring:beans>
        <spring:bean id="dataSource" name="dataSource" class="org.enhydra.jdbc.standard.StandardDataSource" destroy-method="shutdown">
            <spring:property name="driverName" value="org.h2.Driver" />
            <spring:property name="url" value="jdbc:h2:tcp://localhost/~/mule" />
            <spring:property name="user" value="sa" />
            <spring:property name="password">
                <spring:value></spring:value>
            </spring:property>
        </spring:bean>

        <spring:bean id="transactionFactory" name="transactionFactory" class="org.mule.transport.jdbc.JdbcTransactionFactory" />

    </spring:beans>

    <jdbc:connector name="dbConnector" dataSource-ref="dataSource" validateConnections="true" queryTimeout="-1" pollingFrequency="0" doc:name="Database" />

    <flow name="TriggerTxFlow" doc:name="TriggerTxFlow">
        <http:inbound-endpoint exchange-pattern="request-response" host="localhost" port="8081" doc:name="HTTP" />
        <set-variable variableName="flow" value="#[message.inboundProperties['http.query.params']['flow']]" doc:name="Variable"/>
        <vm:outbound-endpoint exchange-pattern="request-response" path="#[flow]" doc:name="VM" />
    </flow>

    <flow name="case1Flow" doc:name="case1Flow">
        <vm:inbound-endpoint exchange-pattern="request-response" path="case1" doc:name="case1">
            <custom-transaction factory-ref="transactionFactory" action="ALWAYS_BEGIN" timeout="10" />
        </vm:inbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert" queryTimeout="-1" connector-ref="dbConnector" doc:name="insert into test values (1, 'Test 1')">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert" value="insert into test values (1, 'Test 1')" />
        </jdbc:outbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert2" queryTimeout="-1" connector-ref="dbConnector" doc:name="insert into test values (2, 'Test 2')">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert2" value="insert into test values (2, 'Test 2')" />
        </jdbc:outbound-endpoint>
    </flow>

    <flow name="case2Flow" doc:name="case2Flow">
        <vm:inbound-endpoint exchange-pattern="request-response" path="case2" doc:name="case2">
            <custom-transaction factory-ref="transactionFactory" action="ALWAYS_BEGIN" timeout="10" />
        </vm:inbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert" queryTimeout="-1" connector-ref="dbConnector" doc:name="Database">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert" value="insert into test values (1, 'Test 1')" />
        </jdbc:outbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert2" queryTimeout="-1" connector-ref="dbConnector" doc:name="Database">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert2" value="insert into test values (2, 'Test 2')" />
        </jdbc:outbound-endpoint>
        <rollback-exception-strategy doc:name="Rollback Exception Strategy" enableNotifications="false" maxRedeliveryAttempts="0">
            <on-redelivery-attempts-exceeded doc:name="Redelivery exhausted">
                <logger message="========= Inside Exception Strategy =========" level="INFO" doc:name="Logger"/>
            </on-redelivery-attempts-exceeded>
        </rollback-exception-strategy>
    </flow>

    <flow name="case3Flow" doc:name="case3Flow">
        <vm:inbound-endpoint exchange-pattern="request-response" path="case3" doc:name="VM">
            <custom-transaction factory-ref="transactionFactory" action="ALWAYS_BEGIN" timeout="10" />
        </vm:inbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert" queryTimeout="-1" connector-ref="dbConnector" doc:name="Database">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert" value="insert into test values (1, 'Test 1')" />
        </jdbc:outbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert2" queryTimeout="-1" connector-ref="dbConnector" doc:name="Database">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert2" value="insert into test values (2, 'Test 2')" />
        </jdbc:outbound-endpoint>
        <catch-exception-strategy doc:name="Catch Exception Strategy">
            <logger message="========= Inside Exception Strategy =========" level="INFO" doc:name="Logger"/>
        </catch-exception-strategy>
    </flow>

</mule>

Answer 1

案例 1：耶！

案例 2：

为什么不调用回滚异常策略？我需要做什么才能调用它？

根据doc 应该调用该策略。这可能是您这边的配置问题，因为如果我添加：

<rollback-exception-strategy>
    <logger message="---> In Rollback exception strategy!!!" />
</rollback-exception-strategy>

到“TxFlow”，我可以看到控制台中记录的文本。

案例 3：

为什么事务没有回滚？

这就是Catch Exception Strategy的主要目的，来自doc：

确保在发生错误时不会回滚由流处理的事务（即永远不会“回滚”事务以重新尝试处理；Mule 提交事务。）

如何在不同于回滚策略的异常策略中强制回滚

我不认为你可以。您可以尝试从策略中抛出异常，但恐怕当时事务已经提交。

Answer 2

试试这个在 ES 中强制回滚：

<custom-exception-strategy class="..." >
    <rollback-transaction exception-pattern="*"/>
</custom-exception-strategy>