Skip to content

Attempt to close all delegate readers even when some fail #4764

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2008-2022 the original author or authors.
* Copyright 2008-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -63,6 +63,7 @@
* @author David Turanski
* @author Mahmoud Ben Hassine
* @author Parikshit Dutta
* @author Elimelec Burghelea
*/
class TaskletStepExceptionTests {

Expand Down Expand Up @@ -212,8 +213,8 @@ public void close() throws ItemStreamException {

taskletStep.execute(stepExecution);
assertEquals(FAILED, stepExecution.getStatus());
assertTrue(stepExecution.getFailureExceptions().contains(taskletException));
assertTrue(stepExecution.getFailureExceptions().contains(exception));
assertEquals(stepExecution.getFailureExceptions().get(0), taskletException);
assertEquals(stepExecution.getFailureExceptions().get(1).getSuppressed()[0], exception);
assertEquals(2, jobRepository.getUpdateCount());
}

Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2006-2023 the original author or authors.
* Copyright 2006-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -761,7 +761,7 @@ public void close() throws ItemStreamException {
Throwable ex = stepExecution.getFailureExceptions().get(0);

// The original rollback was caused by this one:
assertEquals("Bar", ex.getMessage());
assertEquals("Bar", ex.getSuppressed()[0].getMessage());
}

@Test
Expand Down Expand Up @@ -791,7 +791,7 @@ public void close() throws ItemStreamException {
assertEquals("", msg);
Throwable ex = stepExecution.getFailureExceptions().get(0);
// The original rollback was caused by this one:
assertEquals("Bar", ex.getMessage());
assertEquals("Bar", ex.getSuppressed()[0].getMessage());
}

/**
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2024 the original author or authors.
* Copyright 2024-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand All @@ -15,6 +15,7 @@
*/
package org.springframework.batch.item.support;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

Expand All @@ -27,6 +28,7 @@
* implementation is not thread-safe.
*
* @author Mahmoud Ben Hassine
* @author Elimelec Burghelea
* @param <T> type of objects to read
* @since 5.2
*/
Expand Down Expand Up @@ -79,8 +81,22 @@ public void update(ExecutionContext executionContext) throws ItemStreamException

@Override
public void close() throws ItemStreamException {
List<Exception> exceptions = new ArrayList<>();

for (ItemStreamReader<? extends T> delegate : delegates) {
delegate.close();
try {
delegate.close();
}
catch (Exception e) {
exceptions.add(e);
}
}

if (!exceptions.isEmpty()) {
String message = String.format("Failed to close %d delegate(s) due to exceptions", exceptions.size());
ItemStreamException holder = new ItemStreamException(message);
exceptions.forEach(holder::addSuppressed);
throw holder;
}
}

Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2006-2022 the original author or authors.
* Copyright 2006-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -28,7 +28,7 @@
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*
* @author Elimelec Burghelea
*/
public class CompositeItemStream implements ItemStream {

Expand Down Expand Up @@ -102,13 +102,26 @@ public void update(ExecutionContext executionContext) {
/**
* Broadcast the call to close.
* @throws ItemStreamException thrown if one of the {@link ItemStream}s in the list
* fails to close. This is a sequential operation so all itemStreams in the list after
* the one that failed to close will remain open.
* fails to close.
*/
@Override
public void close() throws ItemStreamException {
List<Exception> exceptions = new ArrayList<>();

for (ItemStream itemStream : streams) {
itemStream.close();
try {
itemStream.close();
}
catch (Exception e) {
exceptions.add(e);
}
}

if (!exceptions.isEmpty()) {
String message = String.format("Failed to close %d delegate(s) due to exceptions", exceptions.size());
ItemStreamException holder = new ItemStreamException(message);
exceptions.forEach(holder::addSuppressed);
throw holder;
}
}

Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2024 the original author or authors.
* Copyright 2024-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand All @@ -17,11 +17,14 @@

import java.util.Arrays;

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamReader;

import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
Expand All @@ -32,6 +35,7 @@
* Test class for {@link CompositeItemReader}.
*
* @author Mahmoud Ben Hassine
* @author Elimelec Burghelea
*/
public class CompositeItemReaderTests {

Expand Down Expand Up @@ -107,4 +111,27 @@ void testCompositeItemReaderClose() {
verify(reader2).close();
}

@Test
void testCompositeItemReaderCloseWithDelegateThatThrowsException() {
// given
ItemStreamReader<String> reader1 = mock();
ItemStreamReader<String> reader2 = mock();
CompositeItemReader<String> compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2));

doThrow(new ItemStreamException("A failure")).when(reader1).close();

// when
try {
compositeItemReader.close();
Assertions.fail("Expected an ItemStreamException");
}
catch (ItemStreamException ignored) {

}

// then
verify(reader1).close();
verify(reader2).close();
}

}
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2006-2022 the original author or authors.
* Copyright 2006-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand All @@ -15,19 +15,25 @@
*/
package org.springframework.batch.item.support;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

import java.util.ArrayList;
import java.util.List;

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamSupport;

import static org.junit.jupiter.api.Assertions.assertEquals;

/**
* @author Dave Syer
*
* @author Elimelec Burghelea
*/
class CompositeItemStreamTests {

Expand Down Expand Up @@ -90,6 +96,40 @@ public void close() {
assertEquals(1, list.size());
}

@Test
void testClose2Delegates() {
ItemStream reader1 = Mockito.mock(ItemStream.class);
ItemStream reader2 = Mockito.mock(ItemStream.class);
manager.register(reader1);
manager.register(reader2);

manager.close();

verify(reader1, times(1)).close();
verify(reader2, times(1)).close();
}

@Test
void testClose2DelegatesThatThrowsException() {
ItemStream reader1 = Mockito.mock(ItemStream.class);
ItemStream reader2 = Mockito.mock(ItemStream.class);
manager.register(reader1);
manager.register(reader2);

doThrow(new ItemStreamException("A failure")).when(reader1).close();

try {
manager.close();
Assertions.fail("Expected an ItemStreamException");
}
catch (ItemStreamException ignored) {

}

verify(reader1, times(1)).close();
verify(reader2, times(1)).close();
}

@Test
void testCloseDoesNotUnregister() {
manager.setStreams(new ItemStream[] { new ItemStreamSupport() {
Expand Down
Loading