junrao commented on code in PR #23928:
URL: https://github.com/apache/kafka/pull/23928#discussion_r4364794578
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java:
##########
@@ -541,36 +543,59 @@ private FetchDataInfo
withTransferFailureHandler(FetchDataInfo fetchDataInfo) {
}
/**
- * Called on the network thread when writing fetched records failed, which
may be a socket or a storage failure.
- * Re-reading the region tells them apart, so it is done on the scheduler
rather than on the network thread, at
- * 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.
+ * Called on the network thread when writing fetched records failed in a
way that may come from a failed read of
+ * the segment. A {@link FileReadException} means that the segment could
not be read, which is reported as a
+ * failure of the log dir. Any other exception may be a socket or a
storage failure, which re-reading the region
+ * tells apart. Both are done on the scheduler rather than on the network
thread, since reporting logs
Review Comment:
For FileReadException, I am not sure that it's worth having
tryStartReadFailureReport/finishReadFailureReport and do them in a scheduler.
If we want avoid the error logging, we could change
LogDirFailureChannel.maybeAddOfflineLogDir() to log the error only when the dir
is not offline already. Then, we can just call
logDirFailureChannel.maybeAddOfflineLogDir() inline.
##########
clients/src/main/java/org/apache/kafka/common/network/TransferableChannel.java:
##########
@@ -45,7 +45,20 @@ public interface TransferableChannel extends
GatheringByteChannel {
* @param position The position within the file at which the transfer is
to begin; must be non-negative
* @param count The maximum number of bytes to be transferred; must be
non-negative
* @return The number of bytes, possibly zero, that were actually
transferred
+ * @throws FileReadException If `fileChannel` cannot be read, other than
because it was closed, and this channel
+ * {@link #reportsFileReadFailures()}
+ * @throws IOException If `fileChannel` cannot be read or this channel
cannot be written to
* @see FileChannel#transferTo(long, long,
java.nio.channels.WritableByteChannel)
*/
long transferFrom(FileChannel fileChannel, long position, long count)
throws IOException;
+
+ /**
+ * @return true if {@link #transferFrom} throws a {@link
FileReadException} when the file cannot be read, so that
+ * any other IOException it throws comes from this channel, except
a
+ * {@link java.nio.channels.ClosedChannelException}, which may
come from either channel. false if it cannot
+ * tell a failed read of the file from a failed write to this
channel, as when one system call does both.
+ */
+ default boolean reportsFileReadFailures() {
Review Comment:
Would distinguishesFileReadFailures be a better name?
##########
clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java:
##########
@@ -325,30 +329,56 @@ public int writeTo(TransferableChannel destChannel, int
offset, int length) thro
long position = start + offset;
int count = Math.min(length, oldSize - offset);
try {
- long newSize = Math.min(channel.size(), end) - start;
+ long newSize = Math.min(fileSize(position), 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))
+ if (transferFailureHandler != null &&
mayBeReadFailure(destChannel, e))
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.
+ * The size of the backing file. Failing to get it counts as failing to
read the file at the given position, where
+ * the transfer was to start.
+ */
+ private long fileSize(long position) throws IOException {
+ try {
+ return channel.size();
+ } catch (ClosedChannelException e) {
Review Comment:
Claude suggested the following.
```
Put the wrap rule in one place. "Rethrow a ClosedChannelException, otherwise
wrap in FileReadException" appears three times: FileRecords.fileSize, and
size() and read() in SslTransportLayer. A static helper keeps the rule in one
place:
// In FileReadException
public static IOException ofReadFailure(long position, IOException e) {
// A file closed concurrently, by a deletion or a shutdown, says nothing
about the medium
return e instanceof ClosedChannelException ? e : new
FileReadException(position, e);
}
...
} catch (IOException e) {
throw FileReadException.ofReadFailure(pos, e);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
