Мы действительно запутались с обработкой ошибок в Spring Integration. Мы используем Boot 2.0.2 и Kotlin.
Внутри @Transformer мы выбрасываем исключение X.
Кроме того, в определении потока Java DSL мы выбрасываем такое же исключение X.
каналы
@Bean(TO_BE_PROCESSED_CHANNEL)
fun toBeProcessed() = PublishSubscribeChannel(defaultExecutor())
Поток
@Configuration
class VideoDescriptorPersistSubflow(
val videoRepository: JdbcVideoRepository,
val ingestRecordRepository: CustomIngestRecordRepository
) {
@Bean
fun videoDescriptorPersistFlow(
toBeProcessed: MessageChannel,
processedVideos: MessageChannel
) =
IntegrationFlows.from(toBeProcessed)
.filter { message: Message<*> -> message.ingestRecordId() != null }
.handle { videoDescriptor: VideoDescriptor, _ -> validateVideoDescriptor(videoDescriptor) }
.handle { videoDescriptor: VideoDescriptor, _ -> videoRepository.persist(videoDescriptor) }
.channel(processedVideos)
.get()
fun validateVideoDescriptor(videoDescriptor: VideoDescriptor): VideoDescriptor {
val errors = VideoDescriptorValidator().validate(videoDescriptor)
if (errors.isNotEmpty()) {
throw VideoMetadataValidationException(errors)
}
return videoDescriptor
}
Позже в errorChannel мы отфильтровываем X и делаем кое-что. В этот момент нам нужно ошибочное сообщение.
Для исключения, созданного @Transformer, есть originalMessage.
Для того, что выброшено из подпотока java DSL, originalMessage имеет значение null.
Мы немного покопались и поняли, что первое завернуто в MessagingExceptionWrapper, тогда как второе завернуто в MessageHandlingException, которое не содержит ссылки на исходное сообщение.
Может ли кто-нибудь помочь нам понять, при каких обстоятельствах Spring Integration использует какие исключения? Документы здесь мало что говорят, или мы не смогли найти ничего подходящего.
ОБНОВЛЕНИЕ: ПЕРЕХОД С PUBSUB НА QUEUE CHANNEL ЗАСТАВЛЯЕТ РАБОТАТЬ...
ОБНОВЛЕНИЕ 2. Следуя совету Гэри, теперь мы используем сообщение payload.failed, которое отлично работает. Однако есть что-то хитрое с originalMessage в ErrorMessage.
