Mule FailedToQueueEventException石英连接

Question

我在石英入站组件下方触发了kafka事件。但这似乎引发SEDA队列异常。

<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>

石英用于触发Kafka使用者流程。在Kafka使用者连接在Java组件中结束之前，控件不会返回到调度程序。 Kafka使用者连接将永远不会终止，因为它处于递归while（true）循环中。偶然地，如果Kafka连接结束，则石英调度程序应重新触发Java组件，以重新打开kafka连接。

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

控件不会返回到时间表

Answer 2

之所以引起该错误，是因为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似乎很小。