Colloquio Spring Batch 5: Partitioning, Chunk e Fault Tolerance
Padroneggia i colloqui Spring Batch 5: 15 domande essenziali su partitioning, elaborazione a chunk e fault tolerance con esempi in Java 21.
Spring Batch 5 rappresenta un pilastro per l'elaborazione di dati massivi nell'ecosistema Spring. I colloqui tecnici valutano la capacità di progettare job robusti, scalabili e tolleranti ai guasti. Padroneggiare partitioning, elaborazione orientata ai chunk e meccanismi di fault tolerance distingue gli sviluppatori senior.
Punto chiave del colloquio
I recruiter testano la comprensione profonda: perché scegliere il partitioning rispetto al remote chunking? Come dimensionare correttamente i chunk? Queste decisioni architetturali rivelano una vera esperienza in produzione.
Architettura fondamentale di Spring Batch 5
Domanda 1: Quali sono i componenti principali di Spring Batch?
L'architettura di Spring Batch si basa su tre livelli distinti: l'applicazione (job e codice di business), il Batch Core (classi runtime per avviare e controllare i job) e l'infrastruttura (reader, writer e servizi comuni come RetryTemplate).
BatchJobConfig.javajava
// Configurazione di un job Spring Batch 5 con Java 21
@Configuration
public class BatchJobConfig {
// JobRepository memorizza i metadati di esecuzione
// Permette il restart e il monitoraggio dei job
private final JobRepository jobRepository;
private final PlatformTransactionManager transactionManager;
public BatchJobConfig(JobRepository jobRepository,
PlatformTransactionManager transactionManager) {
this.jobRepository = jobRepository;
this.transactionManager = transactionManager;
}
// Un Job incapsula il processo batch completo
// Composto da uno o più Step eseguiti sequenzialmente
@Bean
public Job importUserJob(Step processUsersStep, Step cleanupStep) {
return new JobBuilder("importUserJob", jobRepository)
.start(processUsersStep) // Step principale di elaborazione
.next(cleanupStep) // Step di pulizia
.build();
}
// Uno Step rappresenta un'unità di lavoro indipendente
// Due modelli: Tasklet (singola attività) o Chunk (elaborazione iterativa)
@Bean
public Step processUsersStep(ItemReader<UserRecord> reader,
ItemProcessor<UserRecord, User> processor,
ItemWriter<User> writer) {
return new StepBuilder("processUsersStep", jobRepository)
.<UserRecord, User>chunk(100, transactionManager) // Commit ogni 100 elementi
.reader(reader) // Legge i dati sorgente
.processor(processor) // Trasforma ogni elemento
.writer(writer) // Scrive in batch da 100
.build();
}
}Il JobRepository persiste lo stato delle esecuzioni nel database. Questa persistenza permette di riavviare un job fallito esattamente dove si è fermato, senza rielaborare i dati già confermati.
Domanda 2: Qual è la differenza tra Tasklet ed elaborazione a Chunk?
Tasklet esegue un'azione discreta, non ripetitiva: eliminazione di file, chiamata a una stored procedure, invio di email di notifica. Chunk elabora volumi massivi suddividendo i dati in batch gestibili.
CleanupTasklet.javajava
// Tasklet: azione singola senza iterazione
@Component
public class CleanupTasklet implements Tasklet {
private final Path tempDirectory = Path.of("/tmp/batch-work");
@Override
public RepeatStatus execute(StepContribution contribution,
ChunkContext chunkContext) throws Exception {
// Elimina tutti i file temporanei dell'elaborazione
try (var files = Files.walk(tempDirectory)) {
files.filter(Files::isRegularFile)
.forEach(this::deleteQuietly);
}
// FINISHED indica che il tasklet ha completato il lavoro
// CONTINUABLE riavvierebbe l'esecuzione (utile per il polling)
return RepeatStatus.FINISHED;
}
private void deleteQuietly(Path file) {
try {
Files.delete(file);
} catch (IOException e) {
// Logga e continua - non far fallire il job per un file
}
}
}ChunkProcessingConfig.javajava
// Elaborazione a chunk: trattamento ad alto volume
@Configuration
public class ChunkProcessingConfig {
@Bean
public Step processOrdersStep(JobRepository jobRepository,
PlatformTransactionManager transactionManager,
ItemReader<OrderRecord> reader,
ItemProcessor<OrderRecord, ProcessedOrder> processor,
ItemWriter<ProcessedOrder> writer) {
return new StepBuilder("processOrdersStep", jobRepository)
// Chunk da 500: legge 500 elementi, elabora, scrive, poi committa
.<OrderRecord, ProcessedOrder>chunk(500, transactionManager)
.reader(reader)
.processor(processor)
.writer(writer)
// Listener per monitorare il progresso
.listener(new ChunkProgressListener())
.build();
}
}L'elaborazione orientata ai chunk fornisce benefici critici: gestione ottimizzata della memoria (solo il chunk corrente in memoria), transazioni granulari (commit per chunk) e recupero dei guasti all'ultimo chunk confermato.
Approfondimento sull'elaborazione a Chunk
Domanda 3: Come funziona il ciclo di vita di un chunk?
Ogni chunk segue un ciclo preciso: lettura degli elementi uno per uno fino a raggiungere la dimensione configurata, elaborazione individuale di ogni elemento, poi scrittura del gruppo. Una transazione avvolge l'intero chunk.
OrderItemReader.javajava
// ItemReader: legge un elemento alla volta
@StepScope
@Component
public class OrderItemReader implements ItemReader<OrderRecord> {
// @StepScope: nuova istanza per ogni esecuzione di step
// Permette di iniettare parametri dinamici del job
@Value("#{jobParameters['startDate']}")
private LocalDate startDate;
private Iterator<OrderRecord> orderIterator;
@BeforeStep
public void initializeReader(StepExecution stepExecution) {
// Carica i dati all'avvio dello step
List<OrderRecord> orders = fetchOrdersFromDate(startDate);
this.orderIterator = orders.iterator();
}
@Override
public OrderRecord read() {
// Restituisce null per segnalare la fine dei dati
// Spring Batch chiama read() finché non riceve null
if (orderIterator.hasNext()) {
return orderIterator.next();
}
return null; // Fine del dataset
}
private List<OrderRecord> fetchOrdersFromDate(LocalDate date) {
// Recupera dalla sorgente dati
return List.of(); // Implementazione effettiva
}
}OrderItemProcessor.javajava
// ItemProcessor: trasforma ogni elemento individualmente
@Component
public class OrderItemProcessor implements ItemProcessor<OrderRecord, ProcessedOrder> {
private final PricingService pricingService;
private final ValidationService validationService;
public OrderItemProcessor(PricingService pricingService,
ValidationService validationService) {
this.pricingService = pricingService;
this.validationService = validationService;
}
@Override
public ProcessedOrder process(OrderRecord item) {
// Restituire null filtra l'elemento (non sarà scritto)
if (!validationService.isValid(item)) {
return null; // Elemento filtrato
}
// Trasformazione di business
BigDecimal finalPrice = pricingService.calculatePrice(item);
return new ProcessedOrder(
item.orderId(),
item.customerId(),
finalPrice,
LocalDateTime.now()
);
}
}OrderItemWriter.javajava
// ItemWriter: scrive l'intero chunk in una sola operazione
@Component
public class OrderItemWriter implements ItemWriter<ProcessedOrder> {
private final JdbcTemplate jdbcTemplate;
public OrderItemWriter(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}
@Override
public void write(Chunk<? extends ProcessedOrder> chunk) {
// Il chunk contiene tutti gli elementi elaborati
// Scrittura batch per performance ottimizzate
List<? extends ProcessedOrder> items = chunk.getItems();
jdbcTemplate.batchUpdate(
"INSERT INTO processed_orders (order_id, customer_id, final_price, processed_at) VALUES (?, ?, ?, ?)",
items,
items.size(),
(ps, order) -> {
ps.setLong(1, order.orderId());
ps.setLong(2, order.customerId());
ps.setBigDecimal(3, order.finalPrice());
ps.setTimestamp(4, Timestamp.valueOf(order.processedAt()));
}
);
}
}Se si verifica un'eccezione durante l'elaborazione del chunk, la transazione viene fatta rollback. Il job può quindi riprendere da quel chunk usando i metadati memorizzati nel JobRepository.
Domanda 4: Come scegliere la dimensione ottimale del chunk?
La dimensione del chunk impatta direttamente performance e consumo di memoria. Un chunk troppo piccolo moltiplica i commit (overhead). Un chunk troppo grande consuma memoria eccessiva e allunga i rollback in caso di guasto.
ChunkSizingConfig.javajava
// Configurazione dinamica della dimensione del chunk
@Configuration
public class ChunkSizingConfig {
// Default ragionevole per la maggior parte dei casi
private static final int DEFAULT_CHUNK_SIZE = 100;
// Per elementi leggeri (pochi campi)
private static final int LIGHT_ITEMS_CHUNK_SIZE = 500;
// Per elementi pesanti (blob, documenti)
private static final int HEAVY_ITEMS_CHUNK_SIZE = 25;
@Bean
public Step processLightDataStep(JobRepository jobRepository,
PlatformTransactionManager txManager,
ItemReader<LightRecord> reader,
ItemWriter<LightRecord> writer) {
return new StepBuilder("processLightDataStep", jobRepository)
// Elementi leggeri: chunk più grandi per meno commit
.<LightRecord, LightRecord>chunk(LIGHT_ITEMS_CHUNK_SIZE, txManager)
.reader(reader)
.writer(writer)
.build();
}
@Bean
public Step processDocumentsStep(JobRepository jobRepository,
PlatformTransactionManager txManager,
ItemReader<Document> reader,
ItemProcessor<Document, ProcessedDocument> processor,
ItemWriter<ProcessedDocument> writer) {
return new StepBuilder("processDocumentsStep", jobRepository)
// Documenti pesanti: chunk più piccoli per limitare la memoria
.<Document, ProcessedDocument>chunk(HEAVY_ITEMS_CHUNK_SIZE, txManager)
.reader(reader)
.processor(processor)
.writer(writer)
.build();
}
}Regola pratica
Iniziare con 100 elementi per chunk, poi adattare in base alle metriche: tempo di commit, uso della memoria e durata del rollback. Usare listener per monitorare e identificare il valore ottimale.
Partitioning per elaborazione parallela
Domanda 5: Cos'è il partitioning e quando usarlo?
Il partitioning divide un dataset in partizioni indipendenti elaborate in parallelo. Ogni partizione viene eseguita nel proprio thread (locale) o su un worker remoto. Questo approccio moltiplica il throughput senza sacrificare la capacità di restart.
PartitionedJobConfig.javajava
// Configurazione di un job partizionato
@Configuration
public class PartitionedJobConfig {
private final JobRepository jobRepository;
private final PlatformTransactionManager transactionManager;
public PartitionedJobConfig(JobRepository jobRepository,
PlatformTransactionManager transactionManager) {
this.jobRepository = jobRepository;
this.transactionManager = transactionManager;
}
@Bean
public Job partitionedImportJob(Step partitionedStep) {
return new JobBuilder("partitionedImportJob", jobRepository)
.start(partitionedStep)
.build();
}
// Step manager: orchestra le partizioni
@Bean
public Step partitionedStep(Partitioner partitioner,
Step workerStep,
TaskExecutor taskExecutor) {
return new StepBuilder("partitionedStep", jobRepository)
// Divide il lavoro tramite il Partitioner
.partitioner("workerStep", partitioner)
// Step eseguito per ogni partizione
.step(workerStep)
// 8 thread paralleli
.taskExecutor(taskExecutor)
// Numero di partizioni da creare
.gridSize(8)
.build();
}
// TaskExecutor per esecuzione parallela
@Bean
public TaskExecutor batchTaskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(8);
executor.setMaxPoolSize(16);
executor.setQueueCapacity(50);
executor.setThreadNamePrefix("batch-partition-");
executor.initialize();
return executor;
}
}RangePartitioner.javajava
// Partitioner basato su intervalli di ID
@Component
public class RangePartitioner implements Partitioner {
private final JdbcTemplate jdbcTemplate;
public RangePartitioner(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}
@Override
public Map<String, ExecutionContext> partition(int gridSize) {
// Recupera i limiti del dataset
Long minId = jdbcTemplate.queryForObject(
"SELECT MIN(id) FROM orders WHERE status = 'PENDING'", Long.class);
Long maxId = jdbcTemplate.queryForObject(
"SELECT MAX(id) FROM orders WHERE status = 'PENDING'", Long.class);
if (minId == null || maxId == null) {
return Map.of(); // Nessun dato da elaborare
}
// Calcola la dimensione di ogni partizione
long range = (maxId - minId) / gridSize + 1;
Map<String, ExecutionContext> partitions = new HashMap<>();
for (int i = 0; i < gridSize; i++) {
ExecutionContext context = new ExecutionContext();
long start = minId + (i * range);
long end = Math.min(start + range - 1, maxId);
// Ogni partizione riceve i propri limiti
context.putLong("minId", start);
context.putLong("maxId", end);
context.putInt("partitionNumber", i);
partitions.put("partition" + i, context);
}
return partitions;
}
}Il partitioning è adatto a dataset grandi in cui gli elementi sono indipendenti. Le partizioni devono essere bilanciate per evitare che una partizione lenta rallenti l'intero job.
Domanda 6: Qual è la differenza tra partitioning locale e remoto?
Il partitioning locale esegue tutte le partizioni sulla stessa JVM con un thread pool. Il partitioning remoto distribuisce le partizioni su più JVM (worker) tramite middleware di messaggistica.
RemotePartitioningConfig.javajava
// Configurazione di partitioning remoto con messaggistica
@Configuration
public class RemotePartitioningConfig {
@Bean
public Step managerStep(JobRepository jobRepository,
Partitioner partitioner,
MessageChannelPartitionHandler partitionHandler) {
return new StepBuilder("managerStep", jobRepository)
.partitioner("workerStep", partitioner)
// Handler che comunica con i worker remoti
.partitionHandler(partitionHandler)
.build();
}
// PartitionHandler invia gli ExecutionContext ai worker
@Bean
public MessageChannelPartitionHandler partitionHandler(
MessagingTemplate messagingTemplate,
JobExplorer jobExplorer) {
MessageChannelPartitionHandler handler = new MessageChannelPartitionHandler();
handler.setStepName("workerStep");
handler.setGridSize(4);
handler.setMessagingOperations(messagingTemplate);
handler.setJobExplorer(jobExplorer);
// Timeout per attendere il completamento dei worker
handler.setPollInterval(5000L);
return handler;
}
}WorkerConfiguration.javajava
// Configurazione lato worker
@Configuration
public class WorkerConfiguration {
private final JobRepository jobRepository;
private final PlatformTransactionManager transactionManager;
public WorkerConfiguration(JobRepository jobRepository,
PlatformTransactionManager transactionManager) {
this.jobRepository = jobRepository;
this.transactionManager = transactionManager;
}
// Il worker riceve le partizioni ed esegue lo step
@Bean
public Step workerStep(ItemReader<OrderRecord> reader,
ItemProcessor<OrderRecord, ProcessedOrder> processor,
ItemWriter<ProcessedOrder> writer) {
return new StepBuilder("workerStep", jobRepository)
.<OrderRecord, ProcessedOrder>chunk(100, transactionManager)
// Reader configurato con @StepScope per ricevere i parametri di partizione
.reader(reader)
.processor(processor)
.writer(writer)
.build();
}
// Reader che usa i limiti di partizione
@Bean
@StepScope
public JdbcCursorItemReader<OrderRecord> partitionedReader(
DataSource dataSource,
@Value("#{stepExecutionContext['minId']}") Long minId,
@Value("#{stepExecutionContext['maxId']}") Long maxId) {
return new JdbcCursorItemReaderBuilder<OrderRecord>()
.name("partitionedOrderReader")
.dataSource(dataSource)
.sql("SELECT * FROM orders WHERE id BETWEEN ? AND ? AND status = 'PENDING'")
.preparedStatementSetter(ps -> {
ps.setLong(1, minId);
ps.setLong(2, maxId);
})
.rowMapper(new OrderRecordRowMapper())
.build();
}
}Pronto a superare i tuoi colloqui su Spring Boot?
Pratica con i nostri simulatori interattivi, flashcards e test tecnici.
Fault tolerance e recupero degli errori
Domanda 7: Quali meccanismi di fault tolerance offre Spring Batch?
Spring Batch offre tre meccanismi complementari: skip (ignorare gli elementi che falliscono), retry (riprovare automaticamente) e restart (riprendere un job fallito). Questi meccanismi si configurano a livello di step.
FaultTolerantStepConfig.javajava
// Configurazione completa di fault tolerance
@Configuration
public class FaultTolerantStepConfig {
@Bean
public Step faultTolerantStep(JobRepository jobRepository,
PlatformTransactionManager transactionManager,
ItemReader<DataRecord> reader,
ItemProcessor<DataRecord, ProcessedRecord> processor,
ItemWriter<ProcessedRecord> writer,
SkipPolicy customSkipPolicy) {
return new StepBuilder("faultTolerantStep", jobRepository)
.<DataRecord, ProcessedRecord>chunk(100, transactionManager)
.reader(reader)
.processor(processor)
.writer(writer)
// Attiva la modalità fault tolerant
.faultTolerant()
// SKIP: ignora fino a 10 errori di validazione
.skipLimit(10)
.skip(ValidationException.class)
.skip(DataIntegrityViolationException.class)
// Alcuni errori non devono mai essere ignorati
.noSkip(FatalBatchException.class)
// RETRY: ritenta gli errori transitori
.retryLimit(3)
.retry(TransientDataAccessException.class)
.retry(DeadlockLoserDataAccessException.class)
// Backoff esponenziale tra i retry
.backOffPolicy(exponentialBackOffPolicy())
// Listener per loggare gli skip
.listener(skipListener())
.build();
}
@Bean
public BackOffPolicy exponentialBackOffPolicy() {
ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();
policy.setInitialInterval(1000); // 1 secondo
policy.setMultiplier(2.0); // Raddoppia ad ogni retry
policy.setMaxInterval(10000); // Massimo 10 secondi
return policy;
}
@Bean
public SkipListener<DataRecord, ProcessedRecord> skipListener() {
return new SkipListener<>() {
@Override
public void onSkipInRead(Throwable t) {
// Logga l'elemento illeggibile
}
@Override
public void onSkipInProcess(DataRecord item, Throwable t) {
// Logga l'elemento fallito in elaborazione
}
@Override
public void onSkipInWrite(ProcessedRecord item, Throwable t) {
// Logga l'elemento fallito in scrittura
}
};
}
}Retry è adatto agli errori transitori (timeout di rete, deadlock di database). Skip è adatto agli errori sui singoli dati che non devono bloccare l'elaborazione complessiva.
Domanda 8: Come implementare uno SkipPolicy personalizzato?
Uno SkipPolicy personalizzato permette una logica decisionale fine: skip in base al tipo di eccezione, numero di errori o criteri di business specifici.
AdaptiveSkipPolicy.javajava
// SkipPolicy con logica di business avanzata
@Component
public class AdaptiveSkipPolicy implements SkipPolicy {
private static final int MAX_SKIP_COUNT = 100;
private static final double MAX_SKIP_PERCENTAGE = 0.05; // 5% massimo
private final AtomicInteger totalProcessed = new AtomicInteger(0);
private final AtomicInteger skipCount = new AtomicInteger(0);
@Override
public boolean shouldSkip(Throwable exception, long skipCountSoFar) {
// Mai ignorare errori fatali
if (exception instanceof FatalBatchException
|| exception instanceof OutOfMemoryError) {
return false;
}
// Limite assoluto di skip
if (skipCountSoFar >= MAX_SKIP_COUNT) {
return false; // Ferma il job
}
// Limite percentuale
int total = totalProcessed.get();
if (total > 1000) { // Applicare solo dopo il warmup
double skipPercentage = (double) skipCountSoFar / total;
if (skipPercentage > MAX_SKIP_PERCENTAGE) {
return false; // Troppi errori in proporzione
}
}
// Salta errori di validazione e dati
return exception instanceof ValidationException
|| exception instanceof DataFormatException
|| exception instanceof IllegalArgumentException;
}
// Chiamato da un listener per tracciare il progresso
public void incrementProcessed() {
totalProcessed.incrementAndGet();
}
}Domanda 9: Come funziona il restart di un job fallito?
Il JobRepository memorizza lo stato di ogni esecuzione. Al restart, Spring Batch identifica l'ultimo chunk confermato e riprende da quel punto. Gli elementi elaborati con successo non vengono rielaborati.
JobRestartService.javajava
// Servizio di gestione del restart dei job
@Service
public class JobRestartService {
private final JobLauncher jobLauncher;
private final JobExplorer jobExplorer;
private final JobRepository jobRepository;
private final Job importJob;
public JobRestartService(JobLauncher jobLauncher,
JobExplorer jobExplorer,
JobRepository jobRepository,
@Qualifier("importJob") Job importJob) {
this.jobLauncher = jobLauncher;
this.jobExplorer = jobExplorer;
this.jobRepository = jobRepository;
this.importJob = importJob;
}
public JobExecution restartFailedJob(Long jobExecutionId) throws Exception {
// Recupera l'esecuzione fallita
JobExecution failedExecution = jobExplorer.getJobExecution(jobExecutionId);
if (failedExecution == null) {
throw new IllegalArgumentException("Job execution not found: " + jobExecutionId);
}
// Verifica che il job possa essere riavviato
if (!failedExecution.getStatus().equals(BatchStatus.FAILED)) {
throw new IllegalStateException("Only FAILED jobs can be restarted");
}
// Usa gli stessi parametri dell'esecuzione originale
JobParameters originalParams = failedExecution.getJobParameters();
// Rilancia il job - riprende automaticamente dall'ultimo checkpoint
return jobLauncher.run(importJob, originalParams);
}
public List<JobExecution> findRestartableJobs() {
// Elenca tutte le esecuzioni FAILED non ancora riavviate
return jobExplorer.findJobInstancesByJobName(importJob.getName(), 0, 100)
.stream()
.flatMap(instance -> jobExplorer.getJobExecutions(instance).stream())
.filter(exec -> exec.getStatus() == BatchStatus.FAILED)
.filter(this::isRestartable)
.toList();
}
private boolean isRestartable(JobExecution execution) {
// Verifica che non esista un'esecuzione di successo più recente
JobInstance instance = execution.getJobInstance();
return jobExplorer.getJobExecutions(instance).stream()
.noneMatch(exec -> exec.getStatus() == BatchStatus.COMPLETED);
}
}Trabocchetto del colloquio
Un job può essere riavviato solo se i JobParameters sono identici. Modificare un parametro crea una nuova istanza di job, perdendo lo storico del progresso.
Scaling e ottimizzazione
Domanda 10: Quali strategie di scaling sono disponibili?
Spring Batch offre quattro strategie: multi-threaded step (più thread leggono in parallelo), parallel steps (step indipendenti in parallelo), remote chunking (elaborazione distribuita) e partitioning (dati distribuiti).
MultiThreadedStepConfig.javajava
// Step multi-threaded: più thread elaborano lo stesso dataset
@Configuration
public class MultiThreadedStepConfig {
@Bean
public Step multiThreadedStep(JobRepository jobRepository,
PlatformTransactionManager transactionManager,
ItemReader<Record> reader,
ItemProcessor<Record, ProcessedRecord> processor,
ItemWriter<ProcessedRecord> writer,
TaskExecutor taskExecutor) {
return new StepBuilder("multiThreadedStep", jobRepository)
.<Record, ProcessedRecord>chunk(100, transactionManager)
// ATTENZIONE: il reader deve essere thread-safe
.reader(synchronizedReader(reader))
.processor(processor)
.writer(writer)
// 4 thread elaborano i chunk in parallelo
.taskExecutor(taskExecutor)
.throttleLimit(4)
.build();
}
// Wrapper per rendere thread-safe il reader
private ItemReader<Record> synchronizedReader(ItemReader<Record> reader) {
SynchronizedItemStreamReader<Record> syncReader = new SynchronizedItemStreamReader<>();
syncReader.setDelegate((ItemStreamReader<Record>) reader);
return syncReader;
}
}ParallelStepsConfig.javajava
// Esecuzione di step indipendenti in parallelo
@Configuration
public class ParallelStepsConfig {
@Bean
public Job parallelJob(JobRepository jobRepository,
Step loadCustomersStep,
Step loadProductsStep,
Step loadOrdersStep,
Step processDataStep) {
// Flow parallelo: customers e products caricati simultaneamente
Flow loadCustomersFlow = new FlowBuilder<Flow>("loadCustomersFlow")
.start(loadCustomersStep)
.build();
Flow loadProductsFlow = new FlowBuilder<Flow>("loadProductsFlow")
.start(loadProductsStep)
.build();
Flow loadOrdersFlow = new FlowBuilder<Flow>("loadOrdersFlow")
.start(loadOrdersStep)
.build();
// Split esegue i flow in parallelo
return new JobBuilder("parallelJob", jobRepository)
.start(new FlowBuilder<Flow>("parallelLoadFlow")
.split(new SimpleAsyncTaskExecutor())
.add(loadCustomersFlow, loadProductsFlow, loadOrdersFlow)
.build())
// Dopo il caricamento parallelo, elaborazione sequenziale
.next(processDataStep)
.build()
.build();
}
}Il multi-threading è adatto quando il reader può essere sincronizzato. Il partitioning è preferibile per grandi volumi poiché ogni partizione ha il proprio reader senza contesa.
Domanda 11: Come monitorare le performance di un job?
Spring Batch espone metriche tramite listener e JobRepository. L'integrazione con Micrometer permette l'export verso Prometheus, Grafana o altri sistemi di monitoraggio.
BatchMetricsConfig.javajava
// Configurazione di monitoraggio con Micrometer
@Configuration
public class BatchMetricsConfig {
private final MeterRegistry meterRegistry;
public BatchMetricsConfig(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
}
@Bean
public JobExecutionListener metricsJobListener() {
return new JobExecutionListener() {
private Timer.Sample jobTimer;
@Override
public void beforeJob(JobExecution jobExecution) {
// Avvia il timer di durata del job
jobTimer = Timer.start(meterRegistry);
Counter.builder("batch.job.started")
.tag("job", jobExecution.getJobInstance().getJobName())
.register(meterRegistry)
.increment();
}
@Override
public void afterJob(JobExecution jobExecution) {
// Registra la durata totale
jobTimer.stop(Timer.builder("batch.job.duration")
.tag("job", jobExecution.getJobInstance().getJobName())
.tag("status", jobExecution.getStatus().toString())
.register(meterRegistry));
// Counter di job per stato
Counter.builder("batch.job.completed")
.tag("job", jobExecution.getJobInstance().getJobName())
.tag("status", jobExecution.getStatus().toString())
.register(meterRegistry)
.increment();
}
};
}
@Bean
public StepExecutionListener metricsStepListener() {
return new StepExecutionListener() {
@Override
public void afterStep(StepExecution stepExecution) {
String jobName = stepExecution.getJobExecution().getJobInstance().getJobName();
String stepName = stepExecution.getStepName();
// Metriche di throughput
Gauge.builder("batch.step.read.count", stepExecution, StepExecution::getReadCount)
.tag("job", jobName)
.tag("step", stepName)
.register(meterRegistry);
Gauge.builder("batch.step.write.count", stepExecution, StepExecution::getWriteCount)
.tag("job", jobName)
.tag("step", stepName)
.register(meterRegistry);
Gauge.builder("batch.step.skip.count", stepExecution, StepExecution::getSkipCount)
.tag("job", jobName)
.tag("step", stepName)
.register(meterRegistry);
return null;
}
};
}
}Domanda 12: Quali sono le insidie comuni del partitioning?
Gli errori frequenti includono: partizioni sbilanciate (una partizione contiene il 90% dei dati), reader non thread-safe e gestione errata dello stato tra partizioni.
BalancedPartitioner.javajava
// Partitioner che bilancia realmente il carico
@Component
public class BalancedPartitioner implements Partitioner {
private final JdbcTemplate jdbcTemplate;
public BalancedPartitioner(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}
@Override
public Map<String, ExecutionContext> partition(int gridSize) {
// Conta gli elementi totali da elaborare
Integer totalCount = jdbcTemplate.queryForObject(
"SELECT COUNT(*) FROM orders WHERE status = 'PENDING'", Integer.class);
if (totalCount == null || totalCount == 0) {
return Map.of();
}
// Calcola la dimensione target per partizione
int itemsPerPartition = (int) Math.ceil((double) totalCount / gridSize);
Map<String, ExecutionContext> partitions = new HashMap<>();
// Usa OFFSET/LIMIT per partizioni bilanciate
// Più costoso degli intervalli ma garantisce il bilanciamento
for (int i = 0; i < gridSize; i++) {
ExecutionContext context = new ExecutionContext();
context.putInt("offset", i * itemsPerPartition);
context.putInt("limit", itemsPerPartition);
context.putInt("partitionNumber", i);
partitions.put("partition" + i, context);
}
return partitions;
}
}
// OffsetBasedReader.java
// Reader compatibile con partitioning basato su offset
@StepScope
@Component
public class OffsetBasedReader implements ItemReader<OrderRecord>, ItemStream {
private final JdbcTemplate jdbcTemplate;
private Iterator<OrderRecord> iterator;
@Value("#{stepExecutionContext['offset']}")
private int offset;
@Value("#{stepExecutionContext['limit']}")
private int limit;
public OffsetBasedReader(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}
@Override
public void open(ExecutionContext executionContext) {
// Carica esattamente la porzione assegnata a questa partizione
List<OrderRecord> records = jdbcTemplate.query(
"SELECT * FROM orders WHERE status = 'PENDING' ORDER BY id LIMIT ? OFFSET ?",
new OrderRecordRowMapper(),
limit, offset
);
this.iterator = records.iterator();
}
@Override
public OrderRecord read() {
return iterator.hasNext() ? iterator.next() : null;
}
@Override
public void update(ExecutionContext executionContext) {
// Salvataggio dello stato per il restart se necessario
}
@Override
public void close() {
// Pulizia
}
}Domande avanzate per senior
Domanda 13: Come gestire le dipendenze tra job?
Spring Batch non gestisce nativamente le dipendenze tra job. Le soluzioni includono: orchestratori esterni (Airflow, Kubernetes CronJob) o implementazione custom con JobExplorer.
JobDependencyService.javajava
// Gestione delle dipendenze tra job
@Service
public class JobDependencyService {
private final JobExplorer jobExplorer;
private final JobLauncher jobLauncher;
private final Map<String, Job> jobs;
public JobDependencyService(JobExplorer jobExplorer,
JobLauncher jobLauncher,
Map<String, Job> jobs) {
this.jobExplorer = jobExplorer;
this.jobLauncher = jobLauncher;
this.jobs = jobs;
}
public JobExecution runWithDependencies(String jobName,
JobParameters params,
List<String> dependsOn) throws Exception {
// Verifica che tutte le dipendenze siano andate a buon fine
for (String dependency : dependsOn) {
if (!hasSuccessfulExecution(dependency, params)) {
throw new JobExecutionException(
"Dependency not satisfied: " + dependency);
}
}
Job job = jobs.get(jobName);
if (job == null) {
throw new IllegalArgumentException("Unknown job: " + jobName);
}
return jobLauncher.run(job, params);
}
private boolean hasSuccessfulExecution(String jobName, JobParameters params) {
// Cerca un'esecuzione COMPLETED con gli stessi parametri di business
return jobExplorer.findJobInstancesByJobName(jobName, 0, 1)
.stream()
.flatMap(instance -> jobExplorer.getJobExecutions(instance).stream())
.filter(exec -> exec.getStatus() == BatchStatus.COMPLETED)
.anyMatch(exec -> matchesBusinessParams(exec.getJobParameters(), params));
}
private boolean matchesBusinessParams(JobParameters actual, JobParameters expected) {
// Confronta i parametri di business (ignora timestamp di esecuzione)
String actualDate = actual.getString("businessDate");
String expectedDate = expected.getString("businessDate");
return Objects.equals(actualDate, expectedDate);
}
}Domanda 14: Come testare efficacemente un job Spring Batch?
Testare un job Spring Batch richiede un approccio a strati: test unitari per i componenti (reader, processor, writer), test di integrazione per gli step e test end-to-end per i job completi.
OrderProcessorTest.javajava
// Test unitario del processor
@ExtendWith(MockitoExtension.class)
class OrderProcessorTest {
@Mock
private PricingService pricingService;
@Mock
private ValidationService validationService;
@InjectMocks
private OrderItemProcessor processor;
@Test
void shouldProcessValidOrder() {
// Given
OrderRecord input = new OrderRecord(1L, 100L, BigDecimal.TEN);
when(validationService.isValid(input)).thenReturn(true);
when(pricingService.calculatePrice(input)).thenReturn(new BigDecimal("12.50"));
// When
ProcessedOrder result = processor.process(input);
// Then
assertThat(result).isNotNull();
assertThat(result.finalPrice()).isEqualTo(new BigDecimal("12.50"));
}
@Test
void shouldFilterInvalidOrder() {
// Given
OrderRecord input = new OrderRecord(1L, 100L, BigDecimal.TEN);
when(validationService.isValid(input)).thenReturn(false);
// When
ProcessedOrder result = processor.process(input);
// Then - null significa filtrato
assertThat(result).isNull();
verify(pricingService, never()).calculatePrice(any());
}
}ImportJobIntegrationTest.javajava
// Test di integrazione del job completo
@SpringBatchTest
@SpringBootTest
@ActiveProfiles("test")
class ImportJobIntegrationTest {
@Autowired
private JobLauncherTestUtils jobLauncherTestUtils;
@Autowired
private JobRepositoryTestUtils jobRepositoryTestUtils;
@Autowired
private JdbcTemplate jdbcTemplate;
@BeforeEach
void setup() {
// Pulisce i metadati tra i test
jobRepositoryTestUtils.removeJobExecutions();
// Resetta i dati di test
jdbcTemplate.execute("DELETE FROM processed_orders");
jdbcTemplate.execute("DELETE FROM orders");
}
@Test
void shouldCompleteJobSuccessfully() throws Exception {
// Given - dati di test
insertTestOrders(100);
// When
JobParameters params = new JobParametersBuilder()
.addLocalDate("businessDate", LocalDate.now())
.addLong("run.id", System.currentTimeMillis())
.toJobParameters();
JobExecution execution = jobLauncherTestUtils.launchJob(params);
// Then
assertThat(execution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
assertThat(countProcessedOrders()).isEqualTo(100);
}
@Test
void shouldHandleEmptyDataset() throws Exception {
// Given - nessun dato
// When
JobExecution execution = jobLauncherTestUtils.launchJob();
// Then - il job ha successo anche senza dati
assertThat(execution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
}
@Test
void shouldRestartFromFailurePoint() throws Exception {
// Given - simula errore a metà elaborazione
insertTestOrders(100);
insertPoisonOrder(50); // Causa un errore
// When - prima esecuzione fallisce
JobExecution firstExecution = jobLauncherTestUtils.launchJob();
assertThat(firstExecution.getStatus()).isEqualTo(BatchStatus.FAILED);
// Correggi i dati
removePoisonOrder(50);
// When - restart
JobExecution restartExecution = jobLauncherTestUtils.launchJob(
firstExecution.getJobParameters());
// Then - riprende dal punto di fallimento
assertThat(restartExecution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
}
private void insertTestOrders(int count) {
for (int i = 1; i <= count; i++) {
jdbcTemplate.update(
"INSERT INTO orders (id, customer_id, amount, status) VALUES (?, ?, ?, 'PENDING')",
i, i * 10, BigDecimal.valueOf(i * 10));
}
}
private int countProcessedOrders() {
return jdbcTemplate.queryForObject(
"SELECT COUNT(*) FROM processed_orders", Integer.class);
}
}Domanda 15: Come ottimizzare le performance di scrittura su database?
La scrittura diventa spesso il collo di bottiglia. Le ottimizzazioni includono: batch insert JDBC, disattivazione dei vincoli durante il caricamento e uso di tabelle di staging.
OptimizedJdbcWriter.javajava
// Writer ottimizzato per grandi volumi
@Component
public class OptimizedJdbcWriter implements ItemWriter<ProcessedOrder> {
private final JdbcTemplate jdbcTemplate;
private final DataSource dataSource;
public OptimizedJdbcWriter(JdbcTemplate jdbcTemplate, DataSource dataSource) {
this.jdbcTemplate = jdbcTemplate;
this.dataSource = dataSource;
}
@Override
public void write(Chunk<? extends ProcessedOrder> chunk) throws Exception {
List<? extends ProcessedOrder> items = chunk.getItems();
if (items.isEmpty()) {
return;
}
// Usa PreparedStatement con batch
try (Connection connection = dataSource.getConnection();
PreparedStatement ps = connection.prepareStatement(
"INSERT INTO processed_orders (order_id, customer_id, final_price, processed_at) " +
"VALUES (?, ?, ?, ?)")) {
for (ProcessedOrder order : items) {
ps.setLong(1, order.orderId());
ps.setLong(2, order.customerId());
ps.setBigDecimal(3, order.finalPrice());
ps.setTimestamp(4, Timestamp.valueOf(order.processedAt()));
ps.addBatch();
}
// Esegue tutte le insert in un'unica operazione di rete
ps.executeBatch();
}
}
}
// StagingTableWriter.java
// Pattern staging table per volumi molto grandi
@Component
public class StagingTableWriter implements ItemWriter<ProcessedOrder>, StepExecutionListener {
private final JdbcTemplate jdbcTemplate;
private String stagingTable;
public StagingTableWriter(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}
@Override
public void beforeStep(StepExecution stepExecution) {
// Crea una tabella temporanea per questo step
stagingTable = "staging_orders_" + stepExecution.getId();
jdbcTemplate.execute(
"CREATE TEMP TABLE " + stagingTable + " (LIKE processed_orders INCLUDING ALL)");
}
@Override
public void write(Chunk<? extends ProcessedOrder> chunk) {
// Scrive nella staging table (senza vincoli FK)
String sql = "INSERT INTO " + stagingTable +
" (order_id, customer_id, final_price, processed_at) VALUES (?, ?, ?, ?)";
jdbcTemplate.batchUpdate(sql, chunk.getItems(), chunk.size(),
(ps, order) -> {
ps.setLong(1, order.orderId());
ps.setLong(2, order.customerId());
ps.setBigDecimal(3, order.finalPrice());
ps.setTimestamp(4, Timestamp.valueOf(order.processedAt()));
});
}
@Override
public ExitStatus afterStep(StepExecution stepExecution) {
if (stepExecution.getStatus() == BatchStatus.COMPLETED) {
// Bulk copy verso la tabella finale
jdbcTemplate.execute(
"INSERT INTO processed_orders SELECT * FROM " + stagingTable);
}
// Pulisce la staging table
jdbcTemplate.execute("DROP TABLE IF EXISTS " + stagingTable);
return stepExecution.getExitStatus();
}
}Conclusione
Padroneggiare Spring Batch 5 nei colloqui tecnici si basa su una comprensione profonda dei meccanismi interni:
✅ Architettura: Job → Step → Chunk (Reader, Processor, Writer)
✅ Elaborazione a chunk: dimensionamento, ciclo di vita, transazioni
✅ Partitioning: locale vs remoto, bilanciamento delle partizioni
✅ Fault tolerance: skip, retry, restart con policy adeguate
✅ Scaling: multi-threading, parallel steps, remote chunking
✅ Test: unitari, integrazione, end-to-end
✅ Ottimizzazione: batch write, staging table, monitoraggio
Le domande avanzate valutano la capacità di giustificare le scelte architetturali in base al contesto: volume di dati, vincoli di tempo, tolleranza agli errori e infrastruttura disponibile.
Inizia a praticare!
Metti alla prova le tue conoscenze con i nostri simulatori di colloquio e test tecnici.
Tag
#spring batch
#spring boot
#java
#batch processing
#interview questions
Condividi
Articoli correlati
Spring Modulith: Architettura del Monolite Modulare Spiegata
Impara Spring Modulith per costruire monoliti modulari in Java. Architettura, moduli, eventi asincroni e testing con esempi Spring Boot 3.
Colloquio Spring Boot: Propagazione delle Transazioni
Padroneggia la propagazione delle transazioni in Spring Boot: REQUIRED, REQUIRES_NEW, NESTED e altro. 12 domande di colloquio con codice e trappole comuni.
Spring Security 6: Autenticazione JWT Completa
Guida pratica per implementare l'autenticazione JWT con Spring Security 6: configurazione, generazione di token, validazione e best practice di sicurezza.