在 Mule Batch 中处理 ArrayList答案

【问题标题】：Processing ArrayList in Mule Batch在 Mule Batch 中处理 ArrayList
【发布时间】：2014-09-25 17:48:11
【问题描述】：

我正在构建一个用于更新插入记录的聚合过程。我有一个分散收集子流，其中每个路径都查询 Salesforce 环境，将数据映射到一致的输出地图列表中，然后分散收集将消息传递到组合集合转换器。输出是一个 ArrayList，其中包含作为批处理的一部分我需要进行 upsert 的所有值。

问题是当 Arraylist 被传递到 Process 步骤时，Mule 会抛出以下错误：

对象“org.mule.transport.NullPayload”的类型不正确。它的类型必须是“{interface java.lang.Iterable,interface java.util.Iterator,interface org.mule.routing.MessageSequence,interface java.util.Collection}”(java.lang.IllegalArgumentException) (com.mulesoft. module.batch.exception.BatchException）。消息负载的类型：NullPayload

使用 Mule Debugger 我可以看到输入阶段完成时的有效负载不为空，并且是 ArrayList 类型。如果我简化流程，请删除分散聚集，以便输入阶段只是一个查询，而 DataMapper 我仍然收到相同的错误。

如何获取地图列表并使用批处理功能对其进行处理？

输入阶段：

                </processor-chain>
                <processor-chain>
                    <sfdc:query config-ref="Salesforce-B" query="dsql:SELECT Amount,CloseDate,CreatedDate,Id,LeadSource,Name,OwnerId,Primary_Source__c,Probability,StageName FROM Opportunity WHERE lastModifiedDate &gt; #[lastUpdate]" doc:name="B" fetchSize="10"/>
                    <logger message="B has #[payload.size().toString()] Opportunities for upsert" level="INFO" doc:name="Logger"/>
                               <data-mapper:transform config-ref="List_Opportunity__To_Map_B" doc:name="Opp to Sales B"/>

                </processor-chain>
            </scatter-gather>
    <combine-collections-transformer doc:name="Combine Collections"/>

【问题讨论】：

标签： arraylist mule batch-processing esb

【解决方案1】：

我无法真正描绘您的流程，因此附上您的 xml 文件将真正帮助我了解您在此处尝试实现的目标。因此，对于您的情况，我的回答可能并不完全正确。

批处理器能够处理地图列表，下面是一个例子：

   <batch:job name="batchlistofmapsBatch1">
        <batch:input>
            <poll doc:name="Poll">
                <fixed-frequency-scheduler frequency="1" timeUnit="SECONDS"/>
                <set-payload value="#[[['Name':'Andre'],['Name':'Mike']]]" doc:name="Set Payload"/>
            </poll>
        </batch:input>
        <batch:process-records>
            <batch:step name="Batch_Step">
                <logger message="#[message.payload]" level="INFO" doc:name="Logger"/>
            </batch:step>
        </batch:process-records>
    </batch:job>

也就是说，据我所知，SFDC 查询（除非它是单个查询）返回一个消费者迭代器而不是一个数组列表，它可能是在某个地方你正在使用它。我建议你从迭代器生成一个列表，这可以通过使用 iteratorUtils (https://commons.apache.org/proper/commons-collections/javadocs/api-2.1.1/org/apache/commons/collections/IteratorUtils.html) 来完成

【讨论】：

嗨 Andre 在测试中我发现如果我只是将查询用作批处理流可以使用它的输入。不幸的是，这不适用于我们的用例，因为我们需要根据来自 5 个不同 SF 组织的查询结果生成输入。附上我的输入阶段，这样您就可以看到只有两个组织的集合的 POC。问题的基础似乎是 DataMapper 在从连接器转换为 Maps 集合时正在输出一个数组列表。我还没有找到让 DataMapper 以批处理流将接受的任何格式输出的方法。
页面呈现时我看不到整个代码块。它应该以标记开头。如果您看不到它，请告诉我。
嗨，很遗憾，我看不到您放置的任何代码...关于生成地图列表的数据映射器，您应该选择“元素列表”而不是“元素”重新进行映射，这将为您提供地图列表而不是单个地图...

【解决方案2】：

安德烈，感谢您的帮助。我最终通过 Mule Support 打开了一个案例，发现当使用轮询端点收集输入时，它能够成功地传递到处理步骤。

当使用 http 端点并通过 Batch 执行引用触发子流时，或者当负载是 ConsumerIterators 的 ArrayList 时，负载被丢弃并在类型变为 NullPayload 时引发错误。

Mule 确认这是一个错误，并将进行更新。

【讨论】：