diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/common/IOUtils.java b/zookeeper-server/src/main/java/org/apache/zookeeper/common/IOUtils.java index 94de2eddb39..f39545f8db1 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/common/IOUtils.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/common/IOUtils.java @@ -23,6 +23,8 @@ import java.io.InputStream; import java.io.OutputStream; import java.io.PrintStream; +import java.util.Arrays; +import java.util.Collection; import org.slf4j.Logger; /* @@ -52,6 +54,18 @@ public static void closeStream(Closeable stream) { * exceptions added as suppressed exceptions */ public static void closeAll(Closeable... closeables) throws IOException { + closeAll(Arrays.asList(closeables)); + } + + /** + * Closes every non-null object, preserving any {@link IOException} thrown. + * + * @param closeables + * the objects to close, in iteration order + * @throws IOException the first exception thrown while closing, with later + * exceptions added as suppressed exceptions + */ + public static void closeAll(Collection closeables) throws IOException { IOException firstException = null; for (Closeable closeable : closeables) { if (closeable != null) { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java index e14e510c2be..328654cca0e 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java @@ -43,6 +43,7 @@ import org.apache.jute.InputArchive; import org.apache.jute.OutputArchive; import org.apache.jute.Record; +import org.apache.zookeeper.common.IOUtils; import org.apache.zookeeper.server.Request; import org.apache.zookeeper.server.ServerMetrics; import org.apache.zookeeper.server.ServerStats; @@ -264,12 +265,10 @@ public synchronized void rollLog() throws IOException { * @throws IOException */ public synchronized void close() throws IOException { - if (logStream != null) { - logStream.close(); - } - for (FileOutputStream log : streamsToFlush) { - log.close(); - } + List toClose = new ArrayList<>(streamsToFlush.size() + 1); + toClose.add(logStream); + toClose.addAll(streamsToFlush); + IOUtils.closeAll(toClose); } @Override diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/common/IOUtilsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/common/IOUtilsTest.java index ac6ffd7de01..453411309b6 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/common/IOUtilsTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/common/IOUtilsTest.java @@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; +import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -107,4 +108,27 @@ public void testCloseAllDoesNotSuppressAnExceptionOnItself() { assertEquals(List.of(3), closed); } + @Test + public void testCloseAllCollectionPreservesFirstFailureAndSuppressesLaterFailures() { + IOException first = new IOException("first"); + IOException second = new IOException("second"); + List closed = new ArrayList<>(); + List closeables = new ArrayList<>(); + closeables.add(null); + closeables.add(() -> { + closed.add(1); + throw first; + }); + closeables.add(() -> { + closed.add(2); + throw second; + }); + closeables.add(() -> closed.add(3)); + + IOException failure = assertThrows(IOException.class, () -> IOUtils.closeAll(closeables)); + + assertSame(first, failure); + assertArrayEquals(new Throwable[]{second}, failure.getSuppressed()); + assertEquals(List.of(1, 2, 3), closed); + } } diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnLogTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnLogTest.java index 5a8cb02f10a..f47b2cbbbba 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnLogTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnLogTest.java @@ -23,17 +23,24 @@ import static org.hamcrest.core.IsEqual.equalTo; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import java.io.BufferedOutputStream; import java.io.EOFException; import java.io.File; +import java.io.FileOutputStream; import java.io.IOException; import java.io.PrintWriter; +import java.lang.reflect.Field; import java.util.Arrays; import java.util.Comparator; import java.util.HashSet; import java.util.List; import java.util.Objects; +import java.util.Queue; import java.util.Random; import java.util.stream.Collectors; import org.apache.jute.Record; @@ -64,6 +71,33 @@ public class FileTxnLogTest extends ZKTestCase { private static final int KB = 1024; + @SuppressWarnings("unchecked") + @Test + public void testCloseAttemptsEveryStream(@TempDir File tmpDir) throws Exception { + FileTxnLog txnLog = new FileTxnLog(tmpDir); + BufferedOutputStream logStream = mock(BufferedOutputStream.class); + FileOutputStream firstStream = mock(FileOutputStream.class); + FileOutputStream secondStream = mock(FileOutputStream.class); + IOException logStreamFailure = new IOException("log stream"); + IOException queuedStreamFailure = new IOException("queued stream"); + doThrow(logStreamFailure).when(logStream).close(); + doThrow(queuedStreamFailure).when(firstStream).close(); + txnLog.logStream = logStream; + + // Inject close failures directly because streamsToFlush is private. + Field streamsField = FileTxnLog.class.getDeclaredField("streamsToFlush"); + streamsField.setAccessible(true); + Queue streams = (Queue) streamsField.get(txnLog); + streams.add(firstStream); + streams.add(secondStream); + + IOException thrown = assertThrows(IOException.class, txnLog::close); + + assertEquals(logStreamFailure, thrown); + assertArrayEquals(new Throwable[]{queuedStreamFailure}, thrown.getSuppressed()); + verify(secondStream).close(); + } + @Test public void testInvalidPreallocSize() { assertEquals(10 * KB, FilePadding.calculateFileSizeWithPadding(7 * KB, 10 * KB, 0),