Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java (Java)

Code before:
        else
            return fetchData.records.lastBatch()
                .map(RecordBatch::nextOffset)
                .orElse(baseOffset);
    }

    /**
     * Flush this log segment to disk.
     *
     * This method is thread-safe.
     */
    public void flush() throws IOException {
        try {
            LOG_FLUSH_TIMER.time(new Callable<Void>() {
                // lambdas cannot declare a more specific exception type, so we use an anonymous inner class
                @Override
                public Void call() throws IOException {
                    log.flush();
                    offsetIndex().flush();
                    timeIndex().flush();
                    txnIndex.flush();
                    return null;
                }
            });
        } catch (ClosedChannelException e) {
            if (!log.file().exists()) {
                // The log segment file has been moved/deleted
                // as part of a replica movement, so this exception is safe to ignore.
                LOGGER.warn("Caught a ClosedChannelException on a non-existent log segment file '{}'. " +
                        "This is expected during a replica move and is safe to ignore.", log.file().getAbsolutePath());
            } else {
                // The file still exists, so this is an unexpected error.
                // Re-throw it to trigger the normal failure handling.
                throw e;
            }
        } catch (Exception e) {
            if (e instanceof IOException)
                throw (IOException) e;
            else if (e instanceof RuntimeException)
                throw (RuntimeException) e;
            else
                throw new IllegalStateException("Unexpected exception thrown: " + e, e);
        }
    }

    /**
     * Update the directory reference for the log and indices in this segment. This would typically be called after a
     * directory is renamed.
     */
    void updateParentDir(File dir) {
        log.updateParentDir(dir);

Review comment (suggestion):
In `replaceCurrentWithFutureLog()`, we rename the log dir to deleted, close the channel and schedule it to be deleted asynchronously. If we get here, it's possible that the renamed file still exists.

Also, while this fixes the issue with flush, the issue with closed channel could be exposed through read too and therefore cause the same issue of forcing the log directory to be offline.

An alternative is to avoid closing `srcLog` in `replaceCurrentWithFutureLog()`. The segments will be closed when the log is deleted after a delay. By that time, the expectation is that there won't be any pending flushes, reads, etc on the log. This approach is also consistent with the approach in `LogManager.asyncDelete()`.

Code after:
            return fetchData.records.lastBatch()
                .map(RecordBatch::nextOffset)
                .orElse(baseOffset);
    }

    /**
     * Flush this log segment to disk.
     *
     * This method is thread-safe.
     */
    public void flush() throws IOException {
        try {
            LOG_FLUSH_TIMER.time(new Callable<Void>() {
                // lambdas cannot declare a more specific exception type, so we use an anonymous inner class
                @Override
                public Void call() throws IOException {
                    log.flush();
                    offsetIndex().flush();
                    timeIndex().flush();
                    txnIndex.flush();
                    return null;
                }
            });
        } catch (Exception e) {
            if (e instanceof IOException)
                throw (IOException) e;
            else if (e instanceof RuntimeException)
                throw (RuntimeException) e;
            else
                throw new IllegalStateException("Unexpected exception thrown: " + e, e);
        }
    }

    /**
     * Update the directory reference for the log and indices in this segment. This would typically be called after a
     * directory is renamed.
     */
    void updateParentDir(File dir) {
        log.updateParentDir(dir);
        lazyOffsetIndex.updateParentDir(dir);
        lazyTimeIndex.updateParentDir(dir);
        txnIndex.updateParentDir(dir);
    }

    /**
     * Change the suffix for the index and log files for this log segment
     * IOException from this method should be handled by the caller
     */
    public void changeFileSuffixes(String oldSuffix, String newSuffix) throws IOException {
        log.renameTo(new File(Utils.replaceSuffix(log.file().getPath(), oldSuffix, newSuffix)));
        lazyOffsetIndex.renameTo(new File(Utils.replaceSuffix(offsetIndexFile().getPath(), oldSuffix, newSuffix)));