Mule FailedToQueueEventException 石英连接

Question

我有以下石英入站组件来触发 kafka 事件。但它似乎正在抛出 SEDA queue Exception。

<quartz:connector name="myQuartzConnector" validateConnections="true">
    <receiver-threading-profile maxThreadsActive="1"/>
</quartz:connector>

<flow name="quartz-scheduler-kafka-consumer-trigger-flow">
    <quartz:inbound-endpoint jobName="Trigger-Kafka-Consumer-Quartz-Job" repeatInterval="1" responseTimeout="10000" connector-ref="myQuartzConnector" doc:name="Quartz">
        <quartz:event-generator-job/>
    </quartz:inbound-endpoint>
    <component class="org.my.myKafkaCOnsumer" doc:name="Java KafkaConsumer"/>
</flow>

Quartz 用于触发 Kafka 消费流。直到 Kafka 消费者连接在 Java 组件中结束，控制才会返回到调度程序。 Kafka 消费者连接永远不会结束，因为它处于递归 while(true) 循环中。碰巧，如果 Kafka 连接结束，quartz 调度程序应该重新触发重新打开 kafka 连接的 java 组件。

Message               : The queue for 'SEDA Stage quartz-scheduler-kafka-consumer-trigger-flow.stage1' did not accept new event within 30000 MILLISECONDS.
Payload               : {NullPayload}
Payload Type          : org.mule.transport.NullPayload
Element               : null @ message-gateway-profile-update-api:null:null
--------------------------------------------------------------------------------
Root Exception stack trace:
org.mule.api.service.FailedToQueueEventException: The queue for 'SEDA Stage quartz-scheduler-kafka-consumer-trigger-flow.stage1' did not accept new event within 30000 MILLISECONDS.
    at org.mule.processor.SedaStageInterceptingMessageProcessor.enqueue(SedaStageInterceptingMessageProcessor.java:139)
    at org.mule.processor.SedaStageInterceptingMessageProcessor.processNextAsync(SedaStageInterceptingMessageProcessor.java:102)
    at org.mule.processor.AsyncInterceptingMessageProcessor.process(AsyncInterceptingMessageProcessor.java:103)
    at org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor.execute(ExceptionToMessagingExceptionExecutionInterceptor.java:27)

Flow snapshot

Answer 1

该错误是因为 Quartz 端点设置为触发速度快于流处理消息的速度。此流具有默认处理策略queued-asynchronous, which means that the events triggered by the Quartz endpoint are sent to a SEDA queue，然后由流线程在可用时进行处理。 Quartz 端点设置为每 1 ms 重复一次，这是非常低的。在那段时间处理组件的机会很小。当流使用的线程池耗尽时，SEDA 队列开始被填满。当队列中的元素超过默认超时以获取线程执行时，您会收到错误消息。 KB https://help.mulesoft.com/s/article/Error-The-queue-for-SEDA-queue-name-did-not-accept-new-event-within-30000-MILLISECONDS

中描述了此问题

您可以将流处理策略更改为同步以重用来自连接器的队列来执行并避免排队，但repeatInterval 似乎不切实际地小。

附带说明，Quartz connector 很久以前就已被弃用。它已被Poll scope 取代。

Answer 2

控制不会返回到计划中

这是不正确的说法。调度程序不知道之前的流程是如何结束的。事实上，旧流程继续执行，新流程的事件开始。 如果您将 logger 作为调度程序之后的第一个组件放在流的开头，您可以看到它。

实际上，由于您正在收听消息，因此它应该不是调度程序。侦听器应该是流的源头。它应该是这样的：

<flow name="quartz-scheduler-kafka-consumer-trigger-flow">
    <component class="org.my.myKafkaCOnsumer" doc:name="Java KafkaConsumer"/>
</flow>

这里有更多关于多个时间表https://simpleflatservice.com/mule4/Multipleschedules.html