У меня есть поток, определенный в xml, как показано ниже:
<task:executor id="s3ArchivingTaskExecutor" pool-size="10" />
<task:scheduler id="s3ArchivingScheduler" pool-size="10" />
<bean id="s3ArchiveResultMessageStore" class="org.springframework.integration.store.SimpleMessageStore" />
<!-- message store reaper, ultimate timeout = 5 hours -->
<bean id="s3ArchiveResultMessageStoreReaper" class="org.springframework.integration.store.MessageGroupStoreReaper">
<property name="messageGroupStore" ref="s3ArchiveResultMessageStore" />
<property name="timeout" value="18000000" />
</bean>
<!-- run reaper every hour -->
<task:scheduled-tasks scheduler="s3ArchivingScheduler">
<task:scheduled ref="s3ArchiveResultMessageStoreReaper" method="run" fixed-rate="3600000" />
</task:scheduled-tasks>
<int:channel id="enterArchiveS3FileChannel" />
<!-- copy the sendTo parameter to another header so sendTo can be used within this pipeline -->
<int:header-enricher input-channel="enterArchiveS3FileChannel"
output-channel="loadSuccessfullyArchivedFilesAndSplitMessageChannel">
<int:header name="subPipelineSendTo" expression="headers['sendTo']" />
<int:error-channel ref="errorLoggingTerminationChannel" />
</int:header-enricher>
<int:channel id="loadSuccessfullyArchivedFilesAndSplitMessageChannel" />
<int:chain input-channel="loadSuccessfullyArchivedFilesAndSplitMessageChannel"
output-channel="splittedArchiveS3FilesChannel">
<int:service-activator ref="loadTransmittedSuccessfullyFileService"
method="gettransmittedSuccessfullyFiles" />
<!-- after this point, no matter success or fail, messages goes back to archiveS3MessageAggregationChannel -->
<int:header-enricher>
<int:error-channel ref="archiveS3FilesErrorHandlingChannel" />
</int:header-enricher>
<!-- split messages -->
<int:splitter ref="perSignatureIntegrationFileMessageSplitter"
method="splitMessage" />
</int:chain>
<!-- parallel processing (fork) -->
<int:publish-subscribe-channel id="splittedArchiveS3FilesChannel"
task-executor="s3ArchivingTaskExecutor" />
<int:chain input-channel="splittedArchiveS3FilesChannel"
output-channel="archiveS3MessageAggregationChannel">
<int:service-activator ref="archiveSignatureIntegrationFileService"
method="archiveFileForPayload" />
<int:service-activator ref="updateFileToArchivedByS3KeyService"
method="updateFileToArchivedByS3Key" />
<int:header-enricher>
<int:error-channel ref="s3ArchiveJobFailedUpdateJobStatusChannel" />
</int:header-enricher>
</int:chain>
<!-- error handling - copy the corelation id and sequence size -->
<int:channel id="archiveS3FilesErrorHandlingChannel" />
<int:header-enricher input-channel="archiveS3FilesErrorHandlingChannel"
output-channel="archiveS3MessageAggregationChannel">
<int:header name="correlationId" expression="payload.failedMessage.headers['correlationId']" />
<int:header name="sequenceSize" expression="payload.failedMessage.headers['sequenceSize']" />
</int:header-enricher>
<!-- join back with aggregator -->
<int:channel id="archiveS3MessageAggregationChannel" />
<!-- group timeout is sequence size (total amount of messages) * 5 minutes -->
<int:aggregator id="s3ArchiveMessageAggregator"
input-channel="archiveS3MessageAggregationChannel"
output-channel="s3ArchiveMessageAggregatorOutputChannel"
message-store="s3ArchiveResultMessageStore"
ref="archiveS3FilesMessageAggregator"
method="aggregate" />
<int:channel id="s3ArchiveMessageAggregatorOutputChannel" />
<!-- restore sendTo header -->
<int:header-enricher input-channel="s3ArchiveMessageAggregatorOutputChannel"
output-channel="sendToHeaderValueRouterChannel">
<int:header name="sendTo" expression="headers['subPipelineSendTo']" />
</int:header-enricher>
В модульном тесте я обернул этот поток шлюзом и предоставил макет для archiveSignatureIntegrationFileService, который всегда вызывает исключение RuntimeException.Я ожидал, что archiveS3FilesErrorHandlingChannel заберет сообщение, но модульный тест не удался с
java.lang.RuntimeException: test fail
at <PROJECT CLASS PATH>.ArchiveSignatureIntegrationFileServiceImpl.archiveFileForPayload(ArchiveSignatureIntegrationFileServiceImpl.java:60)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:171)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:120)
at org.springframework.integration.handler.support.MessagingMethodInvokerHelper$HandlerMethod.invoke(MessagingMethodInvokerHelper.java:1115)
at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.invokeHandlerMethod(MessagingMethodInvokerHelper.java:624)
at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:491)
at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:362)
at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:106)
at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:93)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:123)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169)
at org.springframework.integration.handler.MessageHandlerChain.handleMessageInternal(MessageHandlerChain.java:110)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:162)
at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:224)
at org.springframework.integration.dispatcher.BroadcastingDispatcher.access$000(BroadcastingDispatcher.java:56)
at org.springframework.integration.dispatcher.BroadcastingDispatcher$1.run(BroadcastingDispatcher.java:204)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:834)
Я думал, что исключение было выдано из канала pub-sub с исполнителем задач, так что должно быть в порядке, но получаетсяЯ был неправ.Я был бы очень признателен всем, кто может любезно объяснить мою ошибку.
Заранее спасибо!
Я пытался превратить enterArchiveS3FileChannel в pub-sub с исполнителем задач, но это не помогло.