WSO2 EI 6.4.0中DSS执行报错时onError序列未触发问题排查

问题原因 #

1. Iterator异步执行的上下文隔离：Iterator会将每个迭代任务作为独立的异步消息流处理，主序列的onError仅负责主流程（如Iterator初始化、Aggregate阶段）的错误，无法捕获Iterator子序列中产生的错误。
2. FORCE_ERROR_ON_SOAP_FAULT的作用限制：该属性仅标记SOAP Fault为错误状态，但如果子序列没有配置对应的错误处理逻辑，错误不会向上传递到主序列的onError。
3. Aggregate的默认行为：Aggregate组件会收集所有迭代响应（包括DSS返回的Fault），并将其视为正常响应处理，不会主动触发错误流程。

解决方法 #

1. 为Iterator子序列配置独立的onError处理 #

直接给Iterator的target序列指定onError属性，让子序列的错误直接触发Failure_Sequence：

<iterate expression="$body//rows" id="BigQID">
    <target>
        <sequence onError="Failure_Sequence">
            <property name="FORCE_ERROR_ON_SOAP_FAULT" scope="default" type="STRING" value="true"/>
            <property expression="//f[1]/v/text()" name="Name" scope="default" type="STRING"/>
            <property expression="//f[2]/v/text()" name="Age" scope="default" type="STRING"/>
            
            <!-- Frame payload for DSS Call -->
            <header name="Action" scope="default" value="urn:data_insert_op"/>
            <payloadFactory media-type="xml">
                <format>
                    <Envelope xmlns="http://www.w3.org/2003/05/soap-envelope">
                        <Body>
                            <data_insert_op xmlns="http://ws.wso2.org/dataservice">
                                <Name>$1</Name>
                                <Age>$2</Age>
                            </data_insert_op>
                        </Body>
                    </Envelope>
                </format>
                <args>
                    <arg evaluator="xml" expression="get-property('Name')"/>
                    <arg evaluator="xml" expression="get-property('Age')"/>
                </args>
            </payloadFactory>
            <!-- calling DSS to push data -->
            <call>
                <endpoint>
                    <http method="post" uri-template="http://localhost:8280/services/test_DSS/data_insert_op">
                        <timeout>
                            <duration>60000</duration>
                        </timeout>
                        <suspendOnFailure>
                            <errorCodes>-1</errorCodes>
                            <initialDuration>0</initialDuration>
                            <progressionFactor>1.0</progressionFactor>
                            <maximumDuration>0</maximumDuration>
                        </suspendOnFailure>
                        <markForSuspension>
                            <retriesBeforeSuspension>0</retriesBeforeSuspension>
                        </markForSuspension>
                    </http>
                </endpoint>
            </call>
        </sequence>
    </target>
</iterate>

2. 在Call后显式检查SOAP Fault并触发错误 #

如果需要更细粒度的控制，可在Call之后添加逻辑，判断响应是否为SOAP Fault并手动抛出错误：

<call>
    <!-- 原endpoint配置 -->
</call>
<!-- 检查响应是否包含SOAP Fault -->
<filter xpath="boolean(//soapenv:Fault)" xmlns:soapenv="http://www.w3.org/2003/05/soap-envelope">
    <then>
        <makefault version="soap12">
            <code value="soapenv:Receiver"/>
            <reason value="DSS数据库执行错误"/>
            <detail expression="//soapenv:Fault"/>
        </makefault>
        <throw faultcode="DATABASE_ERROR" faultreason="DSS插入数据失败"/>
    </then>
</filter>

3. 在Aggregate的onComplete阶段统一处理错误 #

若需在聚合所有响应后统一检查错误，可在onComplete序列中添加判断逻辑：

<aggregate id="BigQID">
    <completeCondition>
        <messageCount max="-1" min="-1"/>
    </completeCondition>
    <onComplete aggregateElementType="root" enclosingElementProperty="result" expression="$body/*[1]">
        <filter xpath="boolean(//soapenv:Fault)" xmlns:soapenv="http://www.w3.org/2003/05/soap-envelope">
            <then>
                <log level="custom">
                    <property name="ERROR" value="存在DSS执行失败的任务"/>
                </log>
                <throw faultcode="AGGREGATION_ERROR" faultreason="检测到DSS执行错误"/>
            </then>
            <else>
                <log level="custom">
                    <property name="ON Aggregate SEQ" value="called***"/>
                    <property expression="fn:count(//*[local-name()='UpdatedRowCount'])" name="*********BigQueryETL-TargetCount:: "/>
                </log>
            </else>
        </filter>
    </onComplete>
</aggregate>

内容的提问来源于stack exchange，提问作者Justin