Spring Integration SFTP入站重命名文件触发ReplyRequiredException报错求助

问题描述 #

作为Spring Integration新手，在实现SFTP文件入站功能时，因系统要求用temp_前缀标识未完成下载的文件（默认.writing仅支持后缀），当前配置可完成文件入站及预期重命名，但触发ReplyRequiredException错误，需确认配置正确性并分析报错原因。

配置代码 #

SftpInboundChannelAdapterSpec integrationFlow = Sftp.inboundAdapter(sftpSessionFactory)
        .filter(chainFilter)
        .localFilename(temp_::concat)
        .remoteDirectory(remoteDirectory)
        .deleteRemoteFiles(false)
        .preserveTimestamp(true)
        .localDirectory(new File(localDirectory));
return IntegrationFlows
        .from(integrationFlow, pc -> pc.poller(pm -> pm.cron(cron).maxMessagesPerPoll(1).errorHandler(new SftpUtils.SftpErrorHandler())))
        .transform((File f) -> f.getName().contains(temp_) ?
                f.renameTo(new File(f.getParent(), f.getName().replace(temp_, ""))) :
                null)
        .handle((transformResult, messageHeaders) -> {
            log.info("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX ({}): {}", localDirectory, transformResult);
            return null;
        })
        .get();

错误日志 #

SftpErrorHandler: Error occured in SFTP ingestion flow, exception: nested exception is org.springframework.integration.handler.ReplyRequiredException: No reply produced by handler 'leapInboundFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0', and its 'requiresReply' property is set to true., failedMessage=GenericMessage [payload=src\main\resources\input\genericfile.name, headers={file_originalFile=src\main\resources\input\genericfile.name, id=293a3e85-cd62-37f7-b701-bbacbf27a83f, file_name=genericfile.name, file_relativePath=genericfile.name, timestamp=1668518160014}]

原因分析 #

1. Handle组件默认强制返回结果：流程中的handle属于ServiceActivator组件，其requiresReply属性默认值为true，当该组件返回null时，直接触发ReplyRequiredException，这是报错的直接原因。
2. Transform逻辑存在不合理性：当前transform返回的是renameTo方法的布尔结果（成功/失败），而非文件对象，后续处理无法直接获取操作后的文件；同时若文件名称不包含temp_前缀，会返回null，也可能触发上游组件的requiresReply校验。

解决方案 #

方案1：修改Handle组件，允许返回null #

通过requiresReply(false)配置关闭ServiceActivator的返回结果强制要求：

.handle((transformResult, messageHeaders) -> {
    log.info("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX ({}): {}", localDirectory, transformResult);
    return null;
}, e -> e.requiresReply(false))

方案2：优化Transform逻辑，返回文件对象 #

调整transform逻辑，重命名成功后返回新的文件对象，失败时抛出异常（或根据业务需求处理），保证流程中始终有有效payload传递：

.transform((File f) -> {
    if (f.getName().contains(temp_)) {
        File renamedFile = new File(f.getParent(), f.getName().replace(temp_, ""));
        if (f.renameTo(renamedFile)) {
            return renamedFile;
        } else {
            throw new RuntimeException("Failed to rename file: " + f.getAbsolutePath());
        }
    }
    // 若文件无temp_前缀，可根据业务需求返回原文件或抛出异常
    return f;
})

配置正确性确认 #

SFTP入站适配器的配置是正确的：

• localFilename(temp_::concat)实现了下载文件时添加temp_前缀，满足系统对未完成下载文件的标识要求；
• deleteRemoteFiles(false)保留远程文件、preserveTimestamp(true)保留文件原时间戳的配置符合常规需求。

内容的提问来源于stack exchange，提问作者kacpr