KAFKA-16304: Offline the log directory on a disk error in the zero-copy Fetch path

clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java

@@ -29,6 +29,7 @@
 import java.io.IOException;
 import java.io.RandomAccessFile;
 import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
 import java.nio.file.StandardOpenOption;
@@ -41,9 +42,23 @@
  * instance to enable slicing a range of the log records.
  */
 public class FileRecords extends AbstractRecords implements Closeable {
+    // Size of the buffer used to re-read a region that failed to transfer, see readFailure
+    static final int REREAD_BUFFER_SIZE = 32 * 1024;
+
+    /**
+     * Notified when {@link #writeTo} fails, typically to find out with {@link #readFailure} whether the backing file
+     * could not be read. It is called on the thread that failed to write, a network thread, so it must not block.
+     */
+    @FunctionalInterface
+    public interface TransferFailureHandler {
+        void onTransferFailure(FileRecords records, long position, int count, IOException transferFailure);
+    }
+
     private final boolean isSlice;
     private final int start;
     private final int end;
+    // Notified when writing to a channel fails, or null
+    private final TransferFailureHandler transferFailureHandler;
 
     private final Iterable<FileLogInputStream.FileChannelRecordBatch> batches;
 
@@ -66,6 +81,7 @@ public class FileRecords extends AbstractRecords implements Closeable {
         this.start = 0;
         this.end = end;
         this.isSlice = false;
+        this.transferFailureHandler = null;
 
         if (channel.size() > Integer.MAX_VALUE) {
             throw new KafkaException(
@@ -92,13 +108,15 @@ private FileRecords(
         File file,
         FileChannel channel,
         int start,
-        int end
+        int end,
+        TransferFailureHandler transferFailureHandler
     ) {
         this.file = file;
         this.channel = channel;
         this.start = start;
         this.end = end;
         this.isSlice = true;
+        this.transferFailureHandler = transferFailureHandler;
 
         // don't check the file size since this is just a slice view
         this.size = new AtomicInteger(end - start);
@@ -146,7 +164,26 @@ public FileRecords slice(int position, int size) {
         int availableBytes = availableBytes(position, size);
         int startPosition = this.start + position;
 
-        return new FileRecords(file, channel, startPosition, startPosition + availableBytes);
+        return new FileRecords(file, channel, startPosition, startPosition + availableBytes, transferFailureHandler);
+    }
+
+    /**
+     * Return a copy of this slice whose {@link #writeTo} notifies the given handler when writing to a channel fails.
+     * Slices taken from it with {@link #slice} notify the same handler. This instance is not modified.
+     *
+     * @param handler The handler to notify
+     * @return A copy of this slice that notifies the handler
+     * @throws IllegalStateException If this instance is not a slice
+     */
+    public FileRecords withTransferFailureHandler(TransferFailureHandler handler) {
+        if (!isSlice)
+            throw new IllegalStateException("A transfer failure handler can only be set on a slice of " + file);
+        return new FileRecords(file, channel, start, end, Objects.requireNonNull(handler));
+    }
+
+    // Visible for testing
+    public TransferFailureHandler transferFailureHandler() {
+        return transferFailureHandler;
     }
 
     /**
@@ -278,17 +315,61 @@ public int truncateTo(int targetSize) throws IOException {
 
     @Override
     public int writeTo(TransferableChannel destChannel, int offset, int length) throws IOException {
-        long newSize = Math.min(channel.size(), end) - start;
         int oldSize = sizeInBytes();
-        if (newSize < oldSize)
-            throw new KafkaException(String.format(
-                    "Size of FileRecords %s has been truncated during write: old size %d, new size %d",
-                    file.getAbsolutePath(), oldSize, newSize));
-
         long position = start + offset;
         int count = Math.min(length, oldSize - offset);
-        // safe to cast to int since `count` is an int
-        return (int) destChannel.transferFrom(channel, position, count);
+        try {
+            long newSize = Math.min(channel.size(), end) - start;
+            if (newSize < oldSize)
+                throw new KafkaException(String.format(
+                        "Size of FileRecords %s has been truncated during write: old size %d, new size %d",
+                        file.getAbsolutePath(), oldSize, newSize));
+            // safe to cast to int since `count` is an int
+            return (int) destChannel.transferFrom(channel, position, count);
+        } catch (IOException e) {
+            // The transfer reads the file and writes the socket in one call, so a storage failure and a socket
+            // failure both arrive here as a bare IOException. Let the handler find out which one it was. A closed
+            // channel means the file was not read, so there is nothing to find out.
+            if (transferFailureHandler != null && !(e instanceof ClosedChannelException))
+                transferFailureHandler.onTransferFailure(this, position, count, e);
+            throw e;
+        }
+    }
+
+    /**
+     * Read the given region of the backing file, after {@link #writeTo} failed on it, to tell a storage failure from
+     * a socket failure. The whole region is read rather than its first byte because, depending on the platform, the
+     * transfer may fail anywhere within it without reporting progress. Only a failed read of an open channel counts
+     * as a storage failure: the end of the file may be reached because the log was truncated concurrently (e.g. when
+     * the replica becomes a follower), and the channel may have been closed concurrently, by a deletion or a
+     * shutdown; neither says anything about the medium.
+     *
+     * @param position The position of the region in the file
+     * @param count The size of the region
+     * @return The failure of the read, naming the file and the position, or empty if the region could be read or
+     *         nothing can be concluded
+     */
+    public Optional<IOException> readFailure(long position, int count) {
+        ByteBuffer buffer = ByteBuffer.allocate(Math.min(count, REREAD_BUFFER_SIZE));
+        long regionEnd = position + count;
+        long readPosition = position;
+        try {
+            while (readPosition < regionEnd) {
+                buffer.clear();
+                buffer.limit((int) Math.min(buffer.capacity(), regionEnd - readPosition));
+                int bytesRead = channel.read(buffer, readPosition);
+                if (bytesRead <= 0)
+                    return Optional.empty();
+                readPosition += bytesRead;
+            }
+            return Optional.empty();
+        } catch (ClosedChannelException e) {
+            return Optional.empty();
+        } catch (IOException e) {
+            return Optional.of(new IOException("Failed to read " + file.getAbsolutePath() + " at position "
+                    + readPosition + " after failing to write " + count + " bytes from position " + position
+                    + " to a channel", e));
+        }
     }
 
     /**

clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java

@@ -31,10 +31,12 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -51,6 +53,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -72,6 +75,8 @@ public class FileRecordsTest {
             "efgh".getBytes(),
             "ijkl".getBytes()
     };
+    private static final String LOG_FILE_PATH = "/logs/topic-0/00000000000000000000.log";
+
     private FileRecords fileRecords;
 
     @BeforeEach
@@ -519,6 +524,185 @@ public void testBytesLengthOfWriteTo() throws IOException {
         verify(channel).transferFrom(any(), anyLong(), eq((long) size - firstWritten));
     }
 
+    /**
+     * A failed write is reported to the handler with the region being written, and the exception is rethrown as it
+     * arrived: the handler finds out later whether the file could be read.
+     */
+    @Test
+    public void testWriteToNotifiesHandlerAndRethrows() throws IOException {
+        List<List<Object>> notifications = new ArrayList<>();
+        FileRecords records = fileRecords.slice(1, fileRecords.sizeInBytes() - 1)
+            .withTransferFailureHandler(recordingHandler(notifications));
+        IOException socketFailure = new IOException("Broken pipe");
+
+        IOException thrown = assertThrows(IOException.class,
+            () -> records.writeTo(failingChannel(socketFailure), 2, records.sizeInBytes() - 2));
+        assertSame(socketFailure, thrown);
+        assertEquals(List.of(List.of(records, 3L, records.sizeInBytes() - 2, socketFailure)), notifications);
+    }
+
+    /**
+     * A closed channel means the file was not read, so there is nothing for the handler to find out.
+     */
+    @Test
+    public void testWriteToDoesNotNotifyHandlerWhenChannelClosed() throws IOException {
+        List<List<Object>> notifications = new ArrayList<>();
+        FileRecords records = fileRecords.slice(0, fileRecords.sizeInBytes())
+            .withTransferFailureHandler(recordingHandler(notifications));
+        IOException closed = new ClosedChannelException();
+
+        IOException thrown = assertThrows(IOException.class,
+            () -> records.writeTo(failingChannel(closed), 0, records.sizeInBytes()));
+        assertSame(closed, thrown);
+        assertTrue(notifications.isEmpty());
+    }
+
+    /**
+     * Getting the size of the file before the transfer is part of writing to the channel, so its failure is
+     * reported too.
+     */
+    @Test
+    public void testWriteToNotifiesHandlerWhenSizeFails() throws IOException {
+        FileChannel fileChannel = fileChannelOfSize(1024);
+        List<List<Object>> notifications = new ArrayList<>();
+        FileRecords records = new FileRecords(mock(File.class), fileChannel, 1024).slice(0, 1024)
+            .withTransferFailureHandler(recordingHandler(notifications));
+        IOException sizeFailure = new IOException("Input/output error");
+        when(fileChannel.size()).thenThrow(sizeFailure);
+        TransferableChannel channel = mock(TransferableChannel.class);
+
+        IOException thrown = assertThrows(IOException.class, () -> records.writeTo(channel, 0, 1024));
+        assertSame(sizeFailure, thrown);
+        assertEquals(List.of(List.of(records, 0L, 1024, sizeFailure)), notifications);
+        verify(channel, never()).transferFrom(any(), anyLong(), anyLong());
+    }
+
+    @Test
+    public void testTransferFailureHandlerIsKeptBySlices() {
+        FileRecords.TransferFailureHandler handler = (records, position, count, e) -> { };
+        FileRecords slice = fileRecords.slice(0, fileRecords.sizeInBytes());
+        FileRecords records = slice.withTransferFailureHandler(handler);
+
+        assertSame(handler, records.transferFailureHandler());
+        assertSame(handler, records.slice(1, records.sizeInBytes() - 1).transferFailureHandler());
+        assertEquals(slice.sizeInBytes(), records.sizeInBytes());
+        assertNull(slice.transferFailureHandler());
+    }
+
+    /**
+     * Only slices are written to a channel; the records of a whole segment keep growing.
+     */
+    @Test
+    public void testTransferFailureHandlerCannotBeSetOnNonSlice() {
+        assertThrows(IllegalStateException.class,
+            () -> fileRecords.withTransferFailureHandler((records, position, count, e) -> { }));
+    }
+
+    @Test
+    public void testReadFailureIsEmptyWhenRegionIsReadable() {
+        assertEquals(Optional.empty(), fileRecords.readFailure(0, fileRecords.sizeInBytes()));
+    }
+
+    @Test
+    public void testReadFailureAtStartOfRegion() throws IOException {
+        FileChannel fileChannel = fileChannelOfSize(1024);
+        IOException readFailure = new IOException("Input/output error");
+        when(fileChannel.read(any(ByteBuffer.class), anyLong())).thenThrow(readFailure);
+
+        IOException failure = recordsOn(fileChannel, 1024).readFailure(0, 1024).orElseThrow();
+        assertSame(readFailure, failure.getCause());
+        assertTrue(failure.getMessage().contains(LOG_FILE_PATH + " at position 0"));
+    }
+
+    /**
+     * Depending on the platform, the transfer may fail anywhere within the region without reporting progress, so a
+     * failure past the first bytes of the region must be found as well.
+     */
+    @Test
+    public void testReadFailureInsideRegion() throws IOException {
+        int size = 2 * FileRecords.REREAD_BUFFER_SIZE;
+        FileChannel fileChannel = fileChannelOfSize(size);
+        IOException readFailure = new IOException("Input/output error");
+        when(fileChannel.read(any(ByteBuffer.class), anyLong()))
+            .thenAnswer(FileRecordsTest::fillBuffer)
+            .thenThrow(readFailure);
+
+        IOException failure = recordsOn(fileChannel, size).readFailure(0, size).orElseThrow();
+        assertSame(readFailure, failure.getCause());
+        assertTrue(failure.getMessage().contains("at position " + FileRecords.REREAD_BUFFER_SIZE));
+        verify(fileChannel).read(any(ByteBuffer.class), eq(0L));
+        verify(fileChannel).read(any(ByteBuffer.class), eq((long) FileRecords.REREAD_BUFFER_SIZE));
+    }
+
+    /**
+     * The log can be truncated while the region is re-read, for instance when the replica becomes a follower while
+     * a fetch is being served. The end of the file says nothing about the medium.
+     */
+    @Test
+    public void testReadFailureIsEmptyWhenTruncatedDuringRead() throws IOException {
+        int size = 2 * FileRecords.REREAD_BUFFER_SIZE;
+        FileChannel fileChannel = fileChannelOfSize(size);
+        when(fileChannel.read(any(ByteBuffer.class), anyLong()))
+            .thenAnswer(FileRecordsTest::fillBuffer)
+            .thenReturn(-1);
+
+        assertEquals(Optional.empty(), recordsOn(fileChannel, size).readFailure(0, size));
+    }
+
+    /**
+     * A channel closed while the region is re-read, by a deletion of the segment or a shutdown, says nothing about
+     * the medium.
+     */
+    @Test
+    public void testReadFailureIsEmptyWhenChannelClosedDuringRead() throws IOException {
+        FileChannel fileChannel = fileChannelOfSize(1024);
+        when(fileChannel.read(any(ByteBuffer.class), anyLong())).thenThrow(new ClosedChannelException());
+
+        assertEquals(Optional.empty(), recordsOn(fileChannel, 1024).readFailure(0, 1024));
+    }
+
+    /**
+     * A read that makes no progress cannot conclude anything, and must not spin.
+     */
+    @Test
+    public void testReadFailureIsEmptyWhenReadMakesNoProgress() throws IOException {
+        FileChannel fileChannel = fileChannelOfSize(1024);
+        when(fileChannel.read(any(ByteBuffer.class), anyLong())).thenReturn(0);
+
+        assertEquals(Optional.empty(), recordsOn(fileChannel, 1024).readFailure(0, 1024));
+    }
+
+    // Records each notification as a list of the handler's arguments
+    private static FileRecords.TransferFailureHandler recordingHandler(List<List<Object>> notifications) {
+        return (records, position, count, e) -> notifications.add(List.of(records, position, count, e));
+    }
+
+    private static FileChannel fileChannelOfSize(long size) throws IOException {
+        FileChannel fileChannel = mock(FileChannel.class);
+        when(fileChannel.size()).thenReturn(size);
+        return fileChannel;
+    }
+
+    private static FileRecords recordsOn(FileChannel fileChannel, int size) throws IOException {
+        File file = mock(File.class);
+        when(file.getAbsolutePath()).thenReturn(LOG_FILE_PATH);
+        return new FileRecords(file, fileChannel, size);
+    }
+
+    private static TransferableChannel failingChannel(IOException failure) throws IOException {
+        TransferableChannel channel = mock(TransferableChannel.class);
+        when(channel.transferFrom(any(), anyLong(), anyLong())).thenThrow(failure);
+        return channel;
+    }
+
+    // An answer to FileChannel.read(ByteBuffer, long) that fills the buffer
+    private static int fillBuffer(InvocationOnMock invocation) {
+        ByteBuffer buffer = invocation.getArgument(0);
+        int bytesRead = buffer.remaining();
+        buffer.position(buffer.limit());
+        return bytesRead;
+    }
+
     /**
      * Test two conditions:
      * 1. If the target offset equals the base offset of the first batch

storage/src/main/java/org/apache/kafka/storage/internals/log/FetchDataInfo.java

@@ -54,6 +54,13 @@ public FetchDataInfo(LogOffsetMetadata fetchOffsetMetadata,
         this.delayedRemoteStorageFetch = delayedRemoteStorageFetch;
     }
 
+    /**
+     * @return A copy of this instance with the given records
+     */
+    public FetchDataInfo withRecords(Records records) {
+        return new FetchDataInfo(fetchOffsetMetadata, records, firstEntryIncomplete, abortedTransactions, delayedRemoteStorageFetch);
+    }
+
     public static FetchDataInfo empty(long fetchOffset) {
         return new FetchDataInfo(new LogOffsetMetadata(fetchOffset), MemoryRecords.EMPTY);
     }