KAFKA-16304: Offline the log directory on a disk error in the zero-copy Fetch path by cheeeee · Pull Request #23445 · apache/kafka

cheeeee · 2026-09-13T03:43:51Z

The Fetch response is sent zero copy: the records are read from the
segment by FileRecords.writeTo, on the network thread, while the
response is written to the socket. A failed read of the medium and a
client that went away both arrive there as a bare IOException, and
Selector treats every one of them as a disconnect. A disk error on
this path therefore never offlines the log directory, unlike an I/O
error in the LocalLog operations.

This change offlines it, with the same split as elsewhere in the log:
the layer that sees the IOException reports it, and storage decides.

LocalLog.read gives the FileRecords slices it returns a transfer
failure handler.
When the transfer fails, FileRecords.writeTo passes the region it
was writing to the handler and rethrows the original exception
unchanged. A ClosedChannelException is not passed on: the file was
not read.
The handler re-reads the region on the log's scheduler, at most once
at a time per log directory, and calls
LogDirFailureChannel.maybeAddOfflineLogDir only if the read fails,
with the failed transfer attached. As the log cleaner does, the
directory is offlined and the error is not rethrown.

The network thread does no extra I/O on a failed send. The send itself
is logged as a disconnect, at DEBUG; the ERROR comes from
LogDirFailureChannel.

The classification is as accurate as what the JVM reports, and no more.
Nothing below the JVM is second-guessed: a transient media error that
has cleared by the time of the re-read is attributed to the socket, and
a read that fails for a reason other than the medium is treated as a
storage failure, the criterion LocalLog already applies. At the JVM
level, the re-read of the region:

reads completely: the socket was at fault;
reaches the end of the file: the log was truncated concurrently (e.g.
the replica became a follower);
makes no progress: nothing can be concluded;
fails with ClosedChannelException or a subclass: the segment was
closed or deleted concurrently, which says nothing about the medium
(the reasoning of KAFKA-19571);
fails with any other IOException: the log directory is offlined.

The whole region is re-read rather than its first byte because,
depending on the platform, the transfer can fail anywhere within it
without reporting progress: on macOS both sendfile(2) and read(2)
drop the partial progress of a chunk whose read fails, and on Windows
the JDK does not use TransmitFile for a non-blocking socket and falls
back to a mapped buffer or ReadFile. On Linux the transfer fails at
the first unreadable page. The re-read reads at most what the failed
call was about to send. The message text of the exception is not
inspected: it is platform specific.

Once the directory is offline its segments are closed, and the next
Fetch gets KAFKA_STORAGE_ERROR.

Consequences worth stating:

A broker whose only log directory, or whose metadata log directory,
fails on this path now shuts down, as it does today for any other I/O
error in that directory.
The KRaft metadata log is read through LocalLog too. A read failure
while the leader sends it marks the raft log's directory offline, the
next append on the leader fails, and the raft I/O thread's fatal fault
handler stops the node, as it does today for an I/O error in
LocalLog.read on the metadata log.
The checks run on the broker's shared background scheduler (the raft
manager's own scheduler for the metadata log).
KRaft snapshots are sent through UnalignedFileRecords, which this
change does not cover; I will follow up separately.
On TLS listeners a read error does not reach FileRecords today:
SslTransportLayer.transferFrom swallows it and sends a stale buffer
instead (KAFKA-21280, KAFKA-21280: Do not send stale bytes after a failed file read in SslTransportLayer.transferFrom #23832). This change takes effect on TLS
listeners once that is fixed.

Generated-by: Claude (Anthropic)

Reviewers: Jun Rao junrao@gmail.com

github-actions · 2026-09-22T03:22:18Z

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

junrao

@cheeeee : Thanks for the PR. If we can accurately detect disk failure, we'd want to wire the disk failure signal to LogDirFailureChannel.

junrao · 2026-10-01T17:46:52Z

+        try {
+            if (position >= channel.size())
+                return true;
+            return channel.read(ByteBuffer.allocate(1), position) >= 0;


When a replica switches from a leader to a follower, its log could be truncated while a pending replica fetch request is being served. channel.read() returns -1 when the position is less than the channel size. If this happens, this will inaccurately mark the replica's disk as failed. The previous check on the channel size is not 100% safe since truncation happens asynchronously and can happen immediately after the check.

Thank you very much for explaining this. I hadn't taken into account that the log can be truncated while a send is in flight. An end-of-file from the re-read says nothing about the medium, and the size check cannot close the window. The re-read now treats an end-of-file, a read that makes no progress, and a channel closed concurrently as inconclusive, and keeps the original exception (testReadFailureIsEmptyWhenTruncatedDuringRead).

cheeeee · 2026-10-14T03:53:57Z

Test notes, updated for f371064, which addresses the latest review. The runs below, tables, log excerpt and stress
run, are on 0bb8dfc. Since then de4c0d1 limits the re-reads to one at a time per log directory instead of per
segment, 4b1b646 moves the offline check into LogDirFailureChannel, and f371064 finishes a check that was not
scheduled whatever is thrown. On 4b1b646, on the same host: the PLAINTEXT run gives the same result, the directory
offlined, leadership moved to broker 2 and 20000 of 20000 records consumed; with #23832 at 868d825 so does the SSL
run, with 2 pread64 -EIO instead of 3: one on the network thread, one the re-read; and a 20 minute truncation
stress run had 42 truncations, 113 sends failed on the socket and checked, and 0 log directories offlined. On
f371064 the full storage suite and FileRecordsTest pass, apart from DisableRemoteLogOnTopicTest, a tiered
storage test that timed out once and passes 3 times out of 3 when run alone.

End to end on a cluster: one controller and three brokers in Docker on a Linux 6.8 host, with dm-dust loaded into
the host kernel and broker 1's second log directory on it. t1-0 has replicas 1,2,3 with broker 1 as leader and its
replica on dm-dust; 20000 records are produced, the tail of the segment is made unreadable while its head and
indexes stay in the page cache, and a consumer reads the partition from the beginning. sendfile and pread64 are
traced with strace on all broker threads. Base is 03e6606.

	base, PLAINTEXT	this PR, PLAINTEXT	base, SSL	this PR, SSL	this PR + #23832, SSL
`sendfile` calls / of those `-EIO`	7906 / 771	13 / 1	-	-	-
`pread64` `-EIO`	0	1 (the re-read)	169	169	3
broker 1's `dm-dust` log directory	online	offline	online	online	offline
leader of `t1-0` afterwards	1	2	1	1	2
records consumed	0 of 20000, stalled	20000 of 20000	5456, then `Record is corrupt`	5456, then `Record is corrupt`	20000 of 20000

Over PLAINTEXT the first failed send is logged as a disconnect, the region is re-read on the scheduler, and the
directory goes offline; leadership moves to a healthy replica and the consumer reads everything:

DEBUG [SocketServer listenerType=BROKER, nodeId=1] Connection with /172.31.0.6 (...) disconnected
java.io.IOException: Input/output error
ERROR Failed to read /data/bad/logs/t1-0/00000000000000000000.log at position 2867200 after failing to write 5521408
      bytes from position 2867200 to a channel (org.apache.kafka.storage.internals.log.LogDirFailureChannel)
java.io.IOException: Failed to read /data/bad/logs/t1-0/00000000000000000000.log at position 2867200 ...
	at org.apache.kafka.common.record.internal.FileRecords.readFailure(FileRecords.java:369)
	at org.apache.kafka.storage.internals.log.LocalLog.lambda$checkReadAfterTransferFailure$14(LocalLog.java:558)
	at org.apache.kafka.server.util.KafkaScheduler.lambda$schedule$1(KafkaScheduler.java:151)
	Suppressed: java.io.IOException: Input/output error
		at java.base/sun.nio.ch.FileChannelImpl.transferTo0(Native Method)
WARN [ReplicaManager broker=1] Stopping serving replicas in dir /data/bad/logs ... because the log directory has failed.

Over SSL this PR alone changes nothing: SslTransportLayer.transferFrom swallows the read error and sends the
previous contents of its buffer, so the error never reaches FileRecords and the consumer gets a corrupt batch. That
is KAFKA-21280, a separate bug present since 2.5.0, fixed in #23832; with both, SSL behaves like PLAINTEXT.

Leader -> follower truncation under load, this PR, no fault injected: 20 minutes of producing with acks=1,
consumers killed mid-fetch, and cycles of pausing a follower for 6 s, a preferred leader election and a broker
restart. 25 truncations (Truncating to offset) across the brokers; 102 sends on broker 1 failed on the socket and
had their region checked on the scheduler; 0 log directories offlined.

Cost of a failed send, 200 calls each, on the same loaded host. These are not absolute measurements: the numbers
depend on the host, and they are only meant to show the order of magnitude of the difference. On the network thread
the handler only hands the check over; the re-read runs on the scheduler:

region	network thread, base	network thread, this PR	re-read on the scheduler, in page cache	re-read on the scheduler, from disk
64 KiB	8 us	17 us	0.10 ms	0.9 ms (p99 76 ms)
1 MiB	8 us	13 us	0.26 ms	4.7 ms (p99 442 ms)
8 MiB	8 us	12 us	2.3 ms	29 ms (p99 1.8 s)

Only Linux was exercised end to end; other platforms are covered by the unit tests only.

Tests:

FileRecordsTest: the handler is notified with the region and the exception is rethrown unchanged, including when
getting the size of the file fails; no notification for a closed channel; only slices take a handler; the re-read
finds a failure at the start and inside the region, and is inconclusive on end of file, no progress and a closed
channel.
LocalLogTest: a read failure after a failed transfer offlines the log directory, with the failed transfer
attached, and the next read fails with KafkaStorageException; a readable region keeps it online; one re-read at a
time per log directory, across partitions; a check the scheduler does not take does not block the next one; no
check once the directory is offline.
The full clients, storage, raft, server and core test suites on 4b1b646, as a non-root user, on the host
above under a median load average of 72 on 70 cores: all pass. Checkstyle and spotless pass for clients and
storage.

cheeeee · 2026-10-14T03:54:00Z

(Outdated: since 0bb8dfc writeTo rethrows the original exception and the re-read runs on the log's scheduler; see the description.)

Thanks for the review, Jun. The failure now reaches LogDirFailureChannel: LocalLog.read gives the records it
returns a handler that calls maybeAddOfflineLogDir, and writeTo notifies it before throwing
KafkaStorageException, the same way LocalLog.maybeHandleIOException handles any other IOException. The
criterion is the same as there: an IOException from reading the open channel is a storage failure, while an end of
file, a read without progress and a closed channel are inconclusive. The whole region is re-read now, only on the
error path, since on some platforms the transfer can fail past its first byte. On accuracy: the detection can only
be as accurate as what the JVM reports; the description lists the cases at that level and how each is handled.

…py Fetch path Fetched records are read from the segment by FileRecords.writeTo, on the network thread, while the response is written to the socket. A failed read of the medium and a client that went away both arrive there as a bare IOException, which Selector treats as a disconnect, so a disk error on this path never offlined the log directory. LocalLog.read now gives the FileRecords it returns a read failure handler that calls LogDirFailureChannel.maybeAddOfflineLogDir. When the transfer fails, writeTo re-reads the region it was transferring. If the region reads, the socket was at fault and the original exception is rethrown. If a read fails, the handler is notified and a KafkaStorageException is thrown, as LocalLog.maybeHandleIOException does for any other IOException. Reaching the end of the file (the log was truncated concurrently, e.g. when the replica became a follower), a read making no progress and a closed channel say nothing about the medium and keep the original exception. Without a handler, writeTo behaves as before. The whole region is re-read rather than its first byte because, depending on the platform, the transfer can fail anywhere within it without reporting progress: on macOS both sendfile(2) and read(2) drop the partial progress of a chunk whose read fails, and on Windows the JDK does not use TransmitFile for a non-blocking socket and falls back to a mapped buffer or ReadFile. Generated-by: Claude (Anthropic)

cheeeee · 2026-10-14T15:20:27Z

I've updated the test notes above with a three-broker run, an SSL run and a truncation stress run, and the description with the cost of the re-read. The SSL run turned up a separate bug, KAFKA-21280: on TLS listeners SslTransportLayer.transferFrom swallows a read error and sends a stale buffer, so this change only takes effect there once that is fixed. I'll open a separate PR for it.

(Update: the PR for KAFKA-21280 is #23832; the cost of a failed send is in the test notes.)

cheeeee · 2026-10-14T15:25:54Z

(Updated for de4c0d1.)
The cost of a failed send is acceptable in my opinion: the network thread only hands the check over to the log's scheduler (about 8 us to 12-17 us per failed send, see the test notes), and the re-read runs there, at most one at a time per log directory.
I do not think we need to go deeper than what the JVM exposes to us.
At that level Linux, macOS and Windows are covered: the whole region is re-read, so it does not matter where within it the platform fails the transfer. Only Linux was exercised end to end.

junrao

@cheeeee : Thanks for the updated PR. A few more comments.

junrao · 2026-10-14T22:48:55Z

+     */
+    @Test
+    public void testWriteToReportsStorageFailureInsideRegion() throws IOException {
+        int size = 64 * 1024;


Could we use 2 * FileRecords.REREAD_BUFFER_SIZE? Otherwise, it's not clear later why the position is half the size.

junrao · 2026-10-14T22:50:57Z

+
+    /**
+     * A channel closed while the region is re-read, by a deletion of the segment, a shutdown or an
+     * interrupt, says nothing about the medium, so the original exception must survive.


We don't interrupt the IO threads to avoid failing a read on the file channel.

Dropped the interrupt case.

junrao · 2026-10-14T23:19:59Z

+            // region: only on the error path, and only when someone handles storage failures. The message
+            // text is not inspected, it is platform specific.
+            if (readFailureHandler != null)
+                reReadRegion(position, count, e);


Claude found the following issue, which seems reasonable.

1. The re-read runs on the network thread (FileRecords.java:325-367). writeTo is called from the SocketServer processor thread, through Selector.attemptWrite. So every socket error on such a slice makes that thread read up to count bytes from disk, in 32 KB chunks. Socket errors include a consumer closing, a rebalance, or a timeout ("Connection reset", "Broken pipe"). count is the rest of the partition's slice, typically up to 1 MB. If the failure happened right at position, the rest of the region may not be in the page cache. Meanwhile every other connection on that processor waits, and a burst of disconnects multiplies the stall. The re-read's result doesn't change what happens to the connection, which closes either way. It only decides whether to mark the log directory offline, and that's already handled asynchronously by the log-dir failure handler. So the re-read can move off the network thread: In writeTo: on an IOException, pass (position, count, e) to the handler and rethrow e unchanged. In LocalLog's handler: schedule the re-read on scheduler (LocalLog.java:83), and call maybeAddOfflineLogDir only if it fails. Run at most one re-read per segment file at a time, so a burst of disconnects doesn't queue up hundreds. This also removes the duplicate logging. Today each affected connection logs a WARN "Unexpected error … closing connection" from Selector in addition to the ERROR from LogDirFailureChannel, because KafkaStorageException isn't an IOException.

Done: writeTo passes the region to the handler and rethrows; LocalLog re-reads it on its scheduler, at most once at a time per segment file, and offlines the directory only if the read fails.

junrao · 2026-10-14T23:20:38Z

+            // socket failure both arrive here as a bare IOException. Tell them apart by re-reading the
+            // region: only on the error path, and only when someone handles storage failures. The message
+            // text is not inspected, it is platform specific.
+            if (readFailureHandler != null)


Claude found the following issue.

2. Skip the re-read when the transfer failed with ClosedChannelException (FileRecords.java:327). That means the socket (or the file) was already closed, so the file was never read and the re-read can't tell you anything. Add if (e instanceof ClosedChannelException) throw e; before the re-read. This is the most common way a send fails on a connection that's already closing.

Done. As in KAFKA-19571, a closed channel comes from a concurrent close, not the medium.

junrao · 2026-10-14T23:23:21Z

    }

+    @Test
+    public void testReadFailureWhileWritingFetchedRecordsMarksLogDirOffline() throws IOException {


Claude reported the following issue.

3. The new LocalLogTest leaks the segment files (LocalLogTest.java:262-274). The test leaves the log directory offline, so tearDown's log.close() hits maybeHandleIOException and throws before closing the segments. The open FileChannel and memory-mapped indexes then stay open for the rest of the test JVM. End the test with log.closeQuietly(), as the other offline tests do (:284, :300).

junrao · 2026-10-14T23:33:24Z

        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 {


Claude reported the following.

channel.size() is outside the new try (FileRecords.java:310). An IOException from it skips the handler. That's rare, but moving it inside the try costs nothing.

junrao · 2026-10-14T23:35:38Z

+     */
+    private void reReadRegion(long position, int count, IOException transferFailure) {
+        ByteBuffer buffer = ByteBuffer.allocate(Math.min(count, REREAD_BUFFER_SIZE));
+        long end = position + count;


This hides the instance level field of the same name. Could we rename it to sth like regionEnd?

junrao · 2026-10-14T23:40:11Z

+    private FetchDataInfo withReadFailureHandler(FetchDataInfo fetchDataInfo) {
+        if (!(fetchDataInfo.records instanceof FileRecords fileRecords))
+            return fetchDataInfo;
+        return new FetchDataInfo(fetchDataInfo.fetchOffsetMetadata,


Claude recommended the following.

The FetchDataInfo rebuild copies every field by hand (LocalLog.java:537-545), so a field added later would be silently dropped. A FetchDataInfo.withRecords(Records) method would keep the copy in one place. The extra allocations per read are negligible.

junrao · 2026-10-14T23:47:48Z

+    }
+
+    /**
+     * Return a slice covering the current records of this instance whose {@link #writeTo} reports a failed read of


This doesn't prevent a non-sliced FileRecords calling this method. In that case, a non-slice is returned.

It returned a slice frozen at the current size; it now rejects a non-slice with IllegalStateException.

Address review comments: - writeTo no longer re-reads the region itself: it passes the failed region to the handler and rethrows the original exception. LocalLog's handler re-reads it on the scheduler, at most once at a time per segment file, and marks the log directory offline only if the read fails, with the failed transfer attached to the read failure. The network thread no longer reads from disk on a failed send, and the failure is no longer logged twice. A check the scheduler does not take does not stop later ones. - Skip the handler when the transfer failed with a closed channel: the file was not read. - Get the size of the file inside the try, so that its failure reaches the handler too. - Only slices take a handler; rename the local that hid the end field; add FetchDataInfo.withRecords instead of copying fields by hand. - Tests: use REREAD_BUFFER_SIZE for the region sizes, drop the interrupt case (I/O threads are not interrupted), close the log quietly after marking its directory offline, and cover one re-read at a time per segment, checks the scheduler does not take, and an offline log directory. Generated-by: Claude (Anthropic)

cheeeee · 2026-10-15T18:26:32Z

Thanks, Jun. Addressed in 0bb8dfc; the description and the test notes are updated.

junrao

@cheeeee : Thanks for the updated PR. A few more comments.

junrao · 2026-10-15T21:51:29Z

    private final boolean isSlice;
    private final int start;
    private final int end;
+    // Notified when writing to a channel fails, or null


channel => network channel?

junrao · 2026-10-15T21:54:13Z

+    }
+
+    /**
+     * Return a copy of this slice whose {@link #writeTo} notifies the given handler when writing to a channel fails.


writing to a channel => writing to a network channel ?

junrao · 2026-10-15T22:00:26Z

+        } 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));


to a channel => to a network channel

junrao · 2026-10-15T22:53:38Z

+    }
+
+    @Test
+    public void testNoReadCheckOnceLogDirIsOffline() throws IOException {


Claude found the following issue.

1. testNoReadCheckOnceLogDirIsOffline takes the shared log offline and leaks its files (LocalLogTest.java:364-378). dir comes from randomPartitionLogDir(tmpDir), so dir.getParent() is tmpDir, which is also the parent of the shared log's logDir. Marking it offline makes tearDown's log.close() throw KafkaStorageException before any segment is closed, and the exception is swallowed. The open FileChannel and index mmaps then stay open for the rest of the test JVM. Either: create otherLog under its own temp directory, e.g. TestUtils.randomPartitionLogDir(TestUtils.tempDirectory()), so the shared log isn't affected; or call log.closeQuietly() in the finally, as testReadFailureAfterTransferFailureMarksLogDirOffline does.

junrao · 2026-10-15T23:01:10Z

+     */
+    private void checkReadAfterTransferFailure(FileRecords records, long position, int count, IOException transferFailure) {
+        String file = records.file().getAbsolutePath();
+        if (logDirFailureChannel.hasOfflineLogDir(parentDir) || !pendingReadChecks.add(file))


Claude recommended the following.

Don't let all re-read checks run at once on the shared scheduler (LocalLog.java:556). scheduler is the broker's KafkaScheduler (background.threads, 10 by default). It also runs ISR expiration, log flush, high-watermark and recovery-point checkpoints, retention and async deletes. Checks are de-duplicated only per segment file, so a disk that fails slowly, with reads blocking for tens of seconds, can queue one check per failing segment across many partitions. That can tie up every scheduler thread and stall those tasks for healthy log dirs too. Fix: allow at most one check per log directory at a time, across all partitions. Marking the directory offline is all-or-nothing anyway, so a second check on the same directory adds nothing.

Done: at most one check at a time per log directory, across partitions, tracked by LogDirFailureChannel, which all logs of the broker share.

junrao · 2026-10-15T23:24:45Z

+     * 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 {


For consistency, should we hook this up to SslTransportLayer.transferFrom() to handle disk IO error during read too? This could be done in a separate followup PR.

No separate hook is needed: SslTransportLayer.transferFrom reads the file within the call that FileRecords.writeTo makes, so a read error there reaches the same handler. What stops it today is KAFKA-21280: once part of the region has been sent, transferFrom swallows the error and sends a stale buffer instead. #23832 fixes that. On the cluster in the test notes, a TLS listener with #23832 offlined the directory and all 20000 records were consumed; without it the consumer failed on a corrupt record.

Thanks for the explanation. Agreed.

Since SslTransportLayer.transferFrom() first reads from file channel and then sends the bytes to the network channel. We can categorize the disk IO error more accurately here. Perhaps we can add a new type of disk related IOException through TransferableChannel.transferFrom()?

Agreed. Now that #23832 rethrows the read error, wrapping it in a dedicated IOException at the fileChannel.read call would let LocalLog take the directory offline on TLS listeners without a re-read, and would also catch a media error that has cleared by the time of the re-read. I'd like to do it in a second PR for KAFKA-16304, after this PR and #23832 are merged:

it changes the same lines of SslTransportLayer.transferFrom as KAFKA-21280: Do not send stale bytes after a failed file read in SslTransportLayer.transferFrom #23832, which fixes a bug present since 2.5.0 and could be backported on its own, so I'd rather keep that PR small;

doing it here would make this PR depend on KAFKA-21280: Do not send stale bytes after a failed file read in SslTransportLayer.transferFrom #23832 in code, while the fix here works on its own over PLAINTEXT, and over TLS once KAFKA-21280: Do not send stale bytes after a failed file read in SslTransportLayer.transferFrom #23832 is in;

over PLAINTEXT sendfile reads and sends in one call and reports a disk error and a socket error as the same IOException, so there is no read to wrap: the re-read stays the way to tell them apart there, and the new type only lets the TLS path skip it.

Would that be ok with you, and does this order of the three PRs work for you: this PR and #23832 first, in either order, then the follow-up?

The plan sounds good to me.

The re-read checks run on the broker's shared scheduler, and a slowly failing disk could take up its threads with one check per segment. Taking a log directory offline is all or nothing, so keep one check at a time per log directory, across partitions, in LogDirFailureChannel. Also close the shared log in testNoReadCheckOnceLogDirIsOffline, whose log directory it takes offline, and say network channel where FileRecords refers to the channel it writes to. Generated-by: Claude (Anthropic)

cheeeee · 2026-10-16T01:29:00Z

Thanks, Jun. Addressed in de4c0d1; the description and the test notes are updated.

junrao

@cheeeee : Thanks for the updated PR. A few more comments.

junrao · 2026-10-16T16:58:58Z

+     */
+    private void checkReadAfterTransferFailure(FileRecords records, long position, int count, IOException transferFailure) {
+        String logDir = parentDir;
+        if (logDirFailureChannel.hasOfflineLogDir(logDir) || !logDirFailureChannel.tryStartReadCheck(logDir))


It's probably better to push the hasOfflineLogDir check inside tryStartReadCheck.

junrao · 2026-10-16T17:07:03Z

+            TestUtils.randomPartitionLogDir(TestUtils.tempDirectory()), config, scheduler);
+        try {
+            FileRecords first = fetchedRecords(firstLog);
+            FileRecords readable = spy(first);


readable => sth like spiedFirst ?

junrao · 2026-10-16T17:29:14Z

+     * 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 {


Thanks for the explanation. Agreed.

Since SslTransportLayer.transferFrom() first reads from file channel and then sends the bytes to the network channel. We can categorize the disk IO error more accurately here. Perhaps we can add a new type of disk related IOException through TransferableChannel.transferFrom()?

Move the offline check into LogDirFailureChannel.tryStartReadCheck, which owns both pieces of state, and rename a spy in LocalLogTest. Generated-by: Claude (Anthropic)

cheeeee · 2026-10-16T23:06:50Z

Thanks, Jun. Addressed in 4b1b646; the description and the test notes are updated.

junrao

@cheeeee : Thanks for the updated PR. A few more minor comments.

junrao · 2026-10-19T16:55:06Z

+    }
+
+    /**
+     * Read the given region of the backing file, after {@link #writeTo} failed on it, to tell a storage failure from


Claude suggested the following optimzation.

1. Re-read only the first chunk (FileRecords.readFailure). With both transports, an IOException only escapes from a transferFrom call that sent nothing: SSL: explicitly, through if (totalBytesWritten > 0) return totalBytesWritten;. Plaintext: sendfile reports partial progress before an error. So the bytes that failed to read are at position, within the first REREAD_BUFFER_SIZE bytes. Reading only that much: cuts the extra disk reads caused by ordinary disconnects, which are most transfer failures; shortens how long a slow or hung disk holds a scheduler thread; keeps the per-directory slot free more often, so a real failure elsewhere in the directory is less likely to be skipped while a harmless check runs. That was the reviewer's finding 1. It's still best effort, but the client retrying the bad offset gives another chance. If you do this, update the Javadoc, which says "the transfer may fail anywhere within it without reporting progress". I haven't checked every fallback path inside the JDK's transferTo. The buffered fallback reports partial progress too, and the mmap one fails with an InternalError rather than an IOException, so it never reaches this handler. A quick check before relying on this would be worthwhile.

Gemini seems to confirm the validity of this optimization.

When transferTo() throws an IOException due to a disk issue (such as a bad sector or read error), the physical defect is located within the very first OS page or disk block (typically 4KB to 8KB) starting at the position provided in that specific method call. It does not mean the error occurred at the exact single byte of position, but it does mean the error is at the immediate start of the requested read chunk. Here is how the underlying mechanics dictate this behavior: 1. Partial Transfers Prevent "Future" Exceptions If you request a large transfer (e.g., 1GB) and the disk has a bad sector 500MB into the file, transferTo will not throw an exception on your initial call. Instead, the underlying OS system call (like Linux sendfile or macOS copyfile) will successfully transfer the healthy data up to the error, and Java will return the number of successfully transferred bytes (e.g., 524288000). You are then expected to loop and call transferTo again with the position updated to the new offset. Only on this subsequent call, where the bad sector is now sitting immediately at the new position, will the OS fail immediately and cause Java to throw the IOException. 2. Disk Block Resolution When the exception is finally thrown, the exact corrupted byte might not be the very first byte at position because storage hardware and operating systems do not read files byte-by-byte. They read in blocks or memory pages (usually 4,096 bytes). If position points to the start of a 4KB block, and the disk hardware encounters a cyclic redundancy check (CRC) failure at the 2,000th byte of that block, the hardware fails the read for the entire block. The OS returns an EIO (Input/Output Error) to Java, which throws the exception. The error is somewhere inside that first block. State of the Transfer upon Exception Because of how OS-level POSIX I/O operates, if a transferTo call throws an IOException due to a disk read error: Previous calls: Any bytes counted in the return values of previous transferTo calls were successfully read and sent to the target. Current call: The target WritableByteChannel received exactly zero bytes from the specific transferTo call that threw the exception. Location: The disk failure is guaranteed to be within the first hardware block (usually within 4KB) of the position argument that caused the crash.

Thanks, Jun. I'd rather keep re-reading the whole region. The premise no longer holds on TLS: with #23832, transferFrom rethrows after partial progress, and in the SSL run in the test notes the read failed at position 2613248 of a transfer of 8388608 bytes from position 0. On macOS, sendfile reads 64 pages (256 KiB) at a time and drops the progress of the chunk whose read fails, so the bad bytes can be up to 256 KiB past position.

Re-reading the whole region does not depend on where each platform fails the transfer, so Linux, macOS and Windows are handled the same way, for any read failure the JVM reports within the region, and it is cheap on all of them: a disconnect usually comes right after the send, so the region is in the page cache. In the test notes we measured 64 KiB, 1 MiB and 8 MiB regions on the test host, a VM on RAID 0 SATA SSDs under load: the re-read took 0.1 to 2.3 ms from the page cache and 0.9 to 29 ms (median) from disk, on the scheduler and at most one at a time per log directory. The numbers depend on the host and should not be taken literally, but their spread shows the order of the cost.

@cheeeee : Thanks for the explanation. We can keep the change as it is.

junrao · 2026-10-19T16:57:57Z

+     * most once at a time per log dir, so that a burst of failed sends or a slow failing disk does not take up the
+     * threads of the scheduler, which other tasks of the broker share.
+     */
+    private void checkReadAfterTransferFailure(FileRecords records, long position, int count, IOException transferFailure) {


Claude suggested the following change.

2. Clear the per-directory flag in one place (LocalLog.checkReadAfterTransferFailure). finishReadCheck is called in three places, and an Error thrown by scheduleOnce skips all of them, leaving the directory unchecked until a restart. An example is OutOfMemoryError: unable to create native thread when the executor starts a worker. Use a try/finally with a hand-off flag instead: if (!logDirFailureChannel.tryStartReadCheck(logDir)) return; boolean handedOff = false; try { ScheduledFuture<?> check = scheduler.scheduleOnce("check-read-after-transfer-failure", () -> { try { records.readFailure(position, count).ifPresent(...); } finally { logDirFailureChannel.finishReadCheck(logDir); } }); // A scheduler that is not running ignores the task handedOff = check != null && !check.isCancelled(); } catch (RuntimeException e) { logger.warn("Could not schedule a read check of {} after a failed transfer", records.file().getAbsolutePath(), e); } finally { if (!handedOff) logDirFailureChannel.finishReadCheck(logDir); }

Done, without the null check: scheduleOnce never returns null. The test now also covers an Error from scheduleOnce.

junrao · 2026-10-19T17:00:35Z

+     * 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


Probably worth adding that this is an absolute file position.

Done, here and in TransferFailureHandler.onTransferFailure.

junrao · 2026-10-19T17:14:41Z

+     * 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 {


The plan sounds good to me.

…thrown An Error thrown by scheduleOnce, e.g. OutOfMemoryError when the executor cannot start a thread, skipped every call to finishReadCheck and stopped the log directory from being checked until a restart. Finish the check in a finally unless the scheduler took it. Also document that the positions passed to the transfer failure handler and to readFailure are absolute file positions. Generated-by: Claude (Anthropic)

cheeeee · 2026-10-20T00:34:17Z

Thanks, Jun. Addressed in f371064; the test notes are updated.

junrao

@cheeeee : Thanks for the updated PR. LGTM

cheeeee · 2026-10-21T02:16:51Z

Jun, this has been a very instructive review for me. Thank you for your time and for the detailed explanations; it's been a pleasure working on this with you.

junrao · 2026-10-21T16:10:59Z

@cheeeee : Thank you for patiently addressing all the review comments. Let me know when you have the PR for optimizing the disk failure detection for SslTransportLayer.transferFrom(). Since most production deployment uses SSL, it's probably important to get that optimization in.

github-actions Bot added triage PRs from the community clients small Small PRs labels Sep 13, 2026

github-actions Bot added the needs-attention label Sep 22, 2026

junrao reviewed Oct 1, 2026

View reviewed changes

github-actions Bot removed needs-attention triage PRs from the community labels Oct 2, 2026

cheeeee force-pushed the KAFKA-16304-disambiguate-zero-copy-send branch from 06a4437 to 0cc3275 Compare October 14, 2026 03:53

cheeeee changed the title ~~KAFKA-16304: disambiguate disk from network failures in the zero-copy Fetch path~~ KAFKA-16304: Offline the log directory on a disk error in the zero-copy Fetch path Oct 14, 2026

github-actions Bot added storage Pull requests that target the storage module and removed small Small PRs labels Oct 14, 2026

cheeeee requested a review from junrao October 14, 2026 03:54

cheeeee force-pushed the KAFKA-16304-disambiguate-zero-copy-send branch from 0cc3275 to 5e38656 Compare October 14, 2026 03:55

cheeeee force-pushed the KAFKA-16304-disambiguate-zero-copy-send branch from 5e38656 to 435ed52 Compare October 14, 2026 03:58

junrao reviewed Oct 14, 2026

View reviewed changes

cheeeee mentioned this pull request Oct 15, 2026

KAFKA-21280: Do not send stale bytes after a failed file read in SslTransportLayer.transferFrom #23832

Merged

junrao reviewed Oct 15, 2026

View reviewed changes

junrao added the ci-approved label Oct 15, 2026

junrao reviewed Oct 16, 2026

View reviewed changes

KAFKA-16304: Do not start a read check on an offline log directory

4b1b646

Move the offline check into LogDirFailureChannel.tryStartReadCheck, which owns both pieces of state, and rename a spy in LocalLogTest. Generated-by: Claude (Anthropic)

junrao reviewed Oct 19, 2026

View reviewed changes

junrao approved these changes Oct 20, 2026

View reviewed changes

junrao merged commit 0d82450 into apache:trunk Oct 20, 2026
24 checks passed

Conversation

cheeeee commented Sep 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Sep 22, 2026

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cheeeee Oct 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cheeeee commented Oct 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cheeeee commented Oct 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cheeeee commented Oct 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cheeeee commented Oct 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cheeeee commented Oct 15, 2026

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

cheeeee commented Sep 13, 2026 •

edited

Loading

cheeeee Oct 14, 2026 •

edited

Loading

cheeeee commented Oct 14, 2026 •

edited

Loading

cheeeee commented Oct 14, 2026 •

edited

Loading

cheeeee commented Oct 14, 2026 •

edited

Loading

cheeeee commented Oct 14, 2026 •

edited

Loading