Event-Driven SFTP Downloads with Spring Integration & Spring Batch
In enterprise architectures, data processing relies heavily on secure file transfers. However, files rarely arrive on an SFTP server at predictable times. Relying on a rigid, scheduled CRON job to process files is an anti-pattern—it either runs too early (missing the file) or too late (delaying business reports).
The robust solution is an Event-Driven Architecture. By combining the reactive polling capabilities of Spring Integration with the heavy-duty data processing engines of Spring Batch, we can build a pipeline that automatically "wakes up" the moment a file hits the SFTP server, downloads it, and safely processes the data into an Oracle Database.
📺 Watch the Complete Implementation Live!
If you prefer visual learning, watch the full step-by-step video tutorial where we build this exact Spring Integration to Spring Batch pipeline from scratch.
1. The Architectural Strategy
Before looking at the code, it is vital to understand why we split this into two frameworks.
- Spring Integration (The Listener): Its sole responsibility is network I/O. It polls the remote server, establishes the SSH connection, downloads the file safely, and deletes the remote copy to prevent duplicate processing.
- Spring Batch (The Worker): Its sole responsibility is data integrity. Once the file is safely on the local disk, Spring Batch takes over to read, process, and write the data using transaction-aware chunks.
This decoupling ensures that network timeouts do not roll back database transactions, and database failures do not leave files locked on the remote server.
2. Project Dependencies (pom.xml)
We are using Java 21 and Spring Boot 3.x. We need the spring-boot-starter-integration and the specific spring-integration-sftp module for the networking layer. For the database and processing layer, we pull in Spring Batch and the Oracle JDBC driver.
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.2.0</version>
<relativePath/>
</parent>
<groupId>com.springjavalab</groupId>
<artifactId>spring-integration-sftp-download</artifactId>
<version>0.0.1-SNAPSHOT</version>
<properties>
<java.version>21</java.version>
</properties>
<dependencies>
<!-- Spring Integration -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-sftp</artifactId>
</dependency>
<!-- Spring Batch & Database -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
<groupId>com.oracle.database.jdbc</groupId>
<artifactId>ojdbc11</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
</dependencies>
</project>
3. Application Properties
Here we define our SFTP credentials, local paths, and database URLs. Notice a critical configuration here: spring.batch.job.enabled=false. By default, Spring Boot attempts to run all Batch Jobs immediately on startup. We must disable this because we only want the job to run after Spring Integration has successfully downloaded a file.
spring.application.name=spring-integration-sftp-download
# SFTP Server Configuration
sftp.host=eu-central-1.sftpcloud.io
sftp.port=22
sftp.user=a402ffddbe8842c4b08d393712fafd5a
sftp.password=txguQ2rTop7QIEhVjVNJdzZgQ3FqKDla
sftp.remote.directory=/remote/sftp/data
sftp.local.directory=F://youtube-code-space//reports
# Oracle Database Configuration
spring.datasource.url=jdbc:oracle:thin:@//192.168.56.1:1521/freepdb1
spring.datasource.username=demo
spring.datasource.password=demo
spring.datasource.driver-class-name=oracle.jdbc.OracleDriver
# Crucial: Disable auto-run so Integration flow can trigger the job manually
spring.batch.job.enabled=false
4. Spring Integration Configuration (The Listener)
This configuration class defines our SFTP connection and the integration flow.
Key Architectural Highlights:
- MGET Command: We use
AbstractRemoteFileOutboundGateway.Command.MGET. This is far more efficient than GET because it can pull multiple files matching our*.csvpattern over a single SSH session. - Option.DELETE: Once the files are securely downloaded to the local directory, this option commands the SFTP server to delete the remote files. This guarantees we do not process the same files twice.
- JobOperator Trigger: Inside the final
.handle()block, we use theJobOperatorto programmatically trigger our Batch Job. We inject the current timestamp into theJobParametersto ensure Spring Batch treats every file drop as a brand new, uniqueJobInstance.
package com.springjavalab.config;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.sshd.sftp.client.SftpClient;
import org.springframework.batch.core.job.Job;
import org.springframework.batch.core.job.parameters.JobParameters;
import org.springframework.batch.core.job.parameters.JobParametersBuilder;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.sftp.dsl.Sftp;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import java.io.File;
import java.util.List;
@Configuration
@RequiredArgsConstructor
@Slf4j
public class SftpConfig {
private final JobOperator jobOperator;
private final Job processFileJob;
@Value("${sftp.host}") private String sftpHost;
@Value("${sftp.port}") private int sftpPort;
@Value("${sftp.user}") private String sftpUser;
@Value("${sftp.password}") private String sftpPassword;
@Value("${sftp.remote.directory}") private String remoteDirectory;
@Value("${sftp.local.directory}") private String localDirectory;
@Bean
public SessionFactory<SftpClient.DirEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost(sftpHost);
factory.setPort(sftpPort);
factory.setUser(sftpUser);
factory.setPassword(sftpPassword);
factory.setAllowUnknownKeys(true);
return factory;
}
@Bean
IntegrationFlow sftpDownloadFlow(SessionFactory<SftpClient.DirEntry> sessionFactory) {
return IntegrationFlow
.fromSupplier(
() -> remoteDirectory + "/*.csv",
e -> e.poller(Pollers.fixedDelay(5000)) // Poll remote server every 5 seconds
).handle(
Sftp.outboundGateway(
sessionFactory,
AbstractRemoteFileOutboundGateway.Command.MGET,
"payload")
.localDirectory(new File(localDirectory))
.options(AbstractRemoteFileOutboundGateway.Option.DELETE)
).handle((payload, headers) -> {
List<File> downloadedFiles = (List<File>) payload;
if (downloadedFiles.isEmpty()) {
log.debug("No new files found.");
return null;
}
// Dynamically pass the local directory to the Batch Job
JobParameters params = new JobParametersBuilder()
.addLong("timestamp", System.currentTimeMillis())
.addString("directory", localDirectory)
.toJobParameters();
try {
log.info("Files downloaded safely. Triggering Batch Job.");
jobOperator.start(processFileJob, params);
} catch (Exception ex) {
log.error("Unable to start batch job", ex);
}
return null;
}).get();
}
}
5. Spring Batch Configuration (The Worker)
Once the integration flow places the files in our local directory, the JobOperator launches this batch job. Because we are managing files, our first step utilizes a Tasklet.
A Tasklet is perfect for file-level operations. It retrieves the dynamic directory path from the JobParameters (passed by the Integration flow), scans the directory, and logs the files ready for processing. (In a complete enterprise setup, this step would be followed by a Chunk-oriented step to actually parse the CSV data into the database).
package com.springjavalab.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.Step;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.infrastructure.repeat.RepeatStatus;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;
import java.io.File;
@Configuration
@Slf4j
public class SimpleBatchConfig {
@Bean
public Step printFileNameStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
return new StepBuilder("printFileNameStep", jobRepository)
.tasklet((contribution, chunkContext) -> {
// Retrieve dynamic parameters passed from Spring Integration
String localDirectory = chunkContext.getStepContext().getJobParameters().get("directory").toString();
File folder = new File(localDirectory);
File[] files = folder.listFiles((dir, name) -> name.toLowerCase().endsWith(".csv"));
if (files == null || files.length == 0) {
log.info("No csv file found in the local directory.");
return RepeatStatus.FINISHED;
}
log.info("Found {} files(s) ready for processing.", files.length);
for (File file : files) {
log.info("Processing file :{} ", file.getName());
// Business logic to import the file to Oracle goes here
}
return RepeatStatus.FINISHED;
}, transactionManager).build();
}
@Bean
public Job processFileJob(JobRepository jobRepository, Step printFileNameStep) {
return new JobBuilder("processFileJob", jobRepository)
.start(printFileNameStep)
.build();
}
}
6. Executing the Application
When you run the main application, you will notice that Spring Batch does not execute immediately. Instead, Spring Integration begins polling the SFTP server. The moment you drop a CSV file into the remote directory, the logs will show the file being downloaded, the remote file being deleted, and the Batch Job instantly waking up to process it.
package com.springjavalab;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class SpringIntegrationSftpDownloadApplication {
public static void main(String[] args) {
SpringApplication.run(SpringIntegrationSftpDownloadApplication.class, args);
}
}
In this tutorial, we used
Option.DELETE to remove the remote file after a successful download. While efficient, some enterprise compliance policies forbid data destruction. Alternatively, you can use Option.RENAME or execute an mv command post-transfer to move the file into a remote /processed archive directory.
Conclusion
By combining Spring Integration and Spring Batch, you have built a highly responsive, self-healing data pipeline. You are no longer reliant on arbitrary CRON schedules. Your application acts intelligently—downloading data the exact second it becomes available, and processing it securely inside transactional boundaries.
🔄 Deepen Your Spring Batch Knowledge
Now that you have mastered SFTP event-driven architectures, explore how to parse, validate, and multi-thread the data once it arrives in your local system.
🧱 Spring Batch Core Components
Dive deeper into the relationship between the JobRepository, JobExecution, and StepExecution metadata tables.
📂 Import Multiple CSVs (MultiResourceItemReader)
Learn how to take the multiple files you just downloaded and bulk insert them into Oracle DB.
🔄 Spring Batch ItemProcessor Example
Learn how to validate and transform your CSV data before it hits the database.
🚫 Skip Policy & Error Handling
Prevent a single bad CSV row from failing your entire batch job by implementing fault-tolerant skip logic.
🔁 Retry Mechanism Configuration
Configure backoff policies to automatically retry database inserts if your Oracle connection times out.
🧵 Multithreaded Processing
Dramatically increase the speed of your file processing by configuring your Steps to execute across parallel threads.