I have a problem with spring integration, i try to read large zip file from ftp. when spring integration download the file, some process not complete. i dont know why, that file still have .writing status and not continue the process
the next process if download proccess complete, file .writing should be deleted but it not happen, because the process stoped.
but it not happened at my local env.
sorry for bad english. hope you got my point.
@Configuration
@IntegrationComponentScan
@ConfigurationProperties("flight.datafeed.ftp.route.management")
@Getter
@Setter
@Slf4j
public class AutomaticRouteConfiguration {
private int mode = 2;
private String host;
private int port;
private String username;
private String password;
private String remoteDirectory;
private long queueFixDelay;
private long queueMaxMessagesPerPoll;
private String directoryInput;
private String directoryUncompress;
private String failedDirectory;
private String filterFileName;
private String metadataStoreName;
@Autowired
private KafkaTemplate<Integer, String> kafkaTemplate;
@Autowired
private RouteManagementTransformer routeManagementTransformer;
@Autowired
private MongoDbMetadataStore metadataStore;
@Value("${flight.datafeed.kafka.topic.routeManagement}")
private String topic;
@Bean
public MongoDbMetadataStore metadataStore(MongoDbFactory factory) {
return new MongoDbMetadataStore(factory, "integrationMetadataStore");
}
@Bean
public DefaultFtpSessionFactory ftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost(host);
sf.setPort(port);
sf.setUsername(username);
sf.setPassword(password);
sf.setClientMode(mode);
return sf;
}
@Bean
public FtpInboundFileSynchronizer sftpInboundFileSynchronizer() {
FtpInboundFileSynchronizer fileSynchronizer = new FtpInboundFileSynchronizer(
ftpSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(false);
fileSynchronizer.setRemoteDirectory("/");
fileSynchronizer.setFilter(new FtpRegexPatternFileListFilter(filterFileName));
return fileSynchronizer;
}
@Bean
@InboundChannelAdapter(channel = "automaticRouteFlow.input",
poller = @Poller(fixedDelay = "${flight.datafeed.ftp.route.management.ftpPoolerDelay}", maxMessagesPerPoll = "2"))
public MessageSource<File> sftpMessageSource() {
log.info("pool to ftp");
FtpInboundFileSynchronizingMessageSource source = new FtpInboundFileSynchronizingMessageSource(
sftpInboundFileSynchronizer());
source.setMaxFetchSize(1);
source.setLocalDirectory(new File(directoryInput));
source.setAutoCreateLocalDirectory(true);
source.setLocalFilter(fileFilters());
return source;
}
@Bean
public IntegrationFlow automaticRouteFlow() {
return p -> p
.transform(unZipTransformer(), e -> e.advice())
.split(new UnZipResultSplitter())
.enrichHeaders(s -> s.headerExpressions(h -> h
.put(FtpConstant.UNCOMPRESSED_DIR, "payload.getParentFile()")
.put(FtpConstant.UNCOMPRESSED_FILE, "payload"))
.header(FtpConstant.FAILED_PATH, failedDirectory)
)
.split(Files.splitter(true), e -> e.advice(fileHandlerAdvice()))
.transform(o -> {
try {
return routeManagementTransformer.transform(o.toString());
} catch (Exception e) {
return new PollSkipAdvice();
}
}
)
.transform(Transformers.toJson())
.filter(o -> !o.toString().equals("{}"))
.channel(MessageChannels.publishSubscribe())
.handle(kafkaMessageHandler(kafkaTemplate, topic),
e -> {
e.advice(directoryHandler());
}
);
}
@Bean
public ExpressionEvaluatingRequestHandlerAdvice fileHandlerAdvice() {
ExpressionEvaluatingRequestHandlerAdvice fileHandlerAdvice = new ExpressionEvaluatingRequestHandlerAdvice();
fileHandlerAdvice
.setOnSuccessExpressionString("headers['" + FtpConstant.UNCOMPRESSED_FILE + "'].delete()");
fileHandlerAdvice.setOnFailureExpressionString(
"headers['" + FtpConstant.FILE_ORIGINAL + "'].renameTo(new java.io.File("
+ "headers['" + FtpConstant.FAILED_PATH + "']"
+ "+headers['" + FtpConstant.FILE_NAME + "']))");
return fileHandlerAdvice;
}
@Bean
public ExpressionEvaluatingRequestHandlerAdvice directoryHandler() {
ExpressionEvaluatingRequestHandlerAdvice fileHandlerAdvice = new ExpressionEvaluatingRequestHandlerAdvice();
fileHandlerAdvice
.setOnSuccessExpressionString("headers['" + FtpConstant.UNCOMPRESSED_DIR + "'].delete()");
return fileHandlerAdvice;
}
private KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandler(
KafkaTemplate<Integer, String> kafkaTemplate, String topic) {
return Kafka
.outboundChannelAdapter(kafkaTemplate)
.headerMapper(new SimpleKafkaHeaderMapper())
.topic(topic).sync(false)
.sendTimeout(queueFixDelay);
}
private FileListFilter fileFilters() {
return new CompositeFileListFilter<>(Arrays.asList(
new FileSystemPersistentAcceptOnceFileListFilter(metadataStore, "ftpFile:"),
new AcceptOnceFileListFilter<>()));
}
private UnZipTransformer unZipTransformer() {
UnZipTransformer unZipTransformer = new UnZipTransformer();
unZipTransformer.setZipResultType(ZipResultType.FILE);
unZipTransformer.setWorkDirectory(new File(directoryUncompress));
unZipTransformer.setDeleteFiles(true);
return unZipTransformer;
}
}
sf.setClientMode(FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE);
. You can also try to set trace logginglog4j.category.org.springframework.integration.file=TRACE
as described in docs