From e80201ab0e1e5174fe46267f001d497df16d7f5f Mon Sep 17 00:00:00 2001 From: dhairya05 Date: Thu, 12 Nov 2026 15:27:34 +0530 Subject: [PATCH 1/2] camel-file: do not assume skip() advanced the whole offset The file resume support positions the body at the recorded offset with a single unchecked call: inputStream.skip(file.getLastOffsetValue()); // GenericFileConverter reader.skip(file.getLastOffsetValue()); reader.skip(offset); // IOHelper.toString InputStream.skip and Reader.skip are both specified to skip up to the requested amount and to return how far they actually got, so the return value is the only thing that says where the stream now is. Discarding it means the body silently starts somewhere other than the offset asked for. The case reachable today is an offset past the end of the file, which happens when the recorded offset is stale because the file was truncated or replaced: the stream ran to EOF and the exchange carried an empty or short body as if that were the file's content, instead of reporting that the offset could not be honoured. The stream now uses InputStream.skipNBytes, which loops and throws EOFException if the stream ends first, and readers go through the new IOHelper.skipFully, which does the same for Reader.skip. --- .../component/file/GenericFileConverter.java | 6 ++-- .../java/org/apache/camel/util/IOHelper.java | 25 ++++++++++++- .../org/apache/camel/util/IOHelperTest.java | 36 +++++++++++++++++++ 3 files changed, 64 insertions(+), 3 deletions(-) diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java index 2c7dd3933f9b9..ebde0a78dad38 100644 --- a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java +++ b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java @@ -124,7 +124,9 @@ public static InputStream genericFileToInputStream(GenericFile file, Exchange LOG.debug("Read file {} (no charset)", f); } InputStream inputStream = IOHelper.toInputStream(f, charset); - inputStream.skip(file.getLastOffsetValue()); + // skipNBytes, not skip: a single skip() is allowed to advance fewer bytes than asked for, which + // would leave the body starting in the middle of the data the offset was meant to skip over + inputStream.skipNBytes(file.getLastOffsetValue()); return inputStream; } @@ -202,7 +204,7 @@ public static Reader genericFileToReader(GenericFile file, Exchange exchange) reader = IOHelper.toReader(f, ExchangeHelper.getCharsetName(exchange)); } - reader.skip(file.getLastOffsetValue()); + IOHelper.skipFully(reader, file.getLastOffsetValue()); return reader; } return null; diff --git a/core/camel-util/src/main/java/org/apache/camel/util/IOHelper.java b/core/camel-util/src/main/java/org/apache/camel/util/IOHelper.java index a660a8a7b02fc..76f21379e063b 100644 --- a/core/camel-util/src/main/java/org/apache/camel/util/IOHelper.java +++ b/core/camel-util/src/main/java/org/apache/camel/util/IOHelper.java @@ -22,6 +22,7 @@ import java.io.BufferedWriter; import java.io.ByteArrayInputStream; import java.io.Closeable; +import java.io.EOFException; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; @@ -140,10 +141,32 @@ public static String toString(BufferedReader reader) throws IOException { return toString(reader, INITIAL_OFFSET); } + /** + * Skips exactly {@code offset} characters of the reader. + *

+ * {@link Reader#skip(long)} is allowed to skip fewer characters than asked for, so a single call leaves the reader + * at an unknown position; this keeps calling it until the whole offset has been skipped. + * + * @param reader the reader to advance + * @param offset the number of characters to skip; zero or negative does nothing + * @throws IOException if the reader fails, or ends before {@code offset} characters were skipped + */ + public static void skipFully(Reader reader, long offset) throws IOException { + long remaining = offset; + while (remaining > 0) { + long skipped = reader.skip(remaining); + if (skipped <= 0) { + throw new EOFException( + "Cannot skip " + offset + " characters: the reader ended after " + (offset - remaining)); + } + remaining -= skipped; + } + } + public static String toString(BufferedReader reader, long offset) throws IOException { StringBuilder sb = new StringBuilder(1024); - reader.skip(offset); + skipFully(reader, offset); char[] buf = new char[1024]; try { diff --git a/core/camel-util/src/test/java/org/apache/camel/util/IOHelperTest.java b/core/camel-util/src/test/java/org/apache/camel/util/IOHelperTest.java index 8e652e88753b2..153bcc08e64b1 100644 --- a/core/camel-util/src/test/java/org/apache/camel/util/IOHelperTest.java +++ b/core/camel-util/src/test/java/org/apache/camel/util/IOHelperTest.java @@ -18,9 +18,13 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.FilterReader; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.io.Reader; +import java.io.StringReader; import java.util.Arrays; import java.util.Locale; @@ -141,4 +145,36 @@ public void testNormalizeEnvironmentVariableTurkishLocale() { } } + @Test + public void testSkipFullyWithAReaderThatSkipsInSmallSteps() throws IOException { + Reader reader = new OneAtATimeSkipReader(new StringReader("0123456789")); + + IOHelper.skipFully(reader, 4); + + assertEquals("456789", IOHelper.toString(reader)); + } + + @Test + public void testSkipFullyPastTheEnd() { + Reader reader = new OneAtATimeSkipReader(new StringReader("0123")); + + assertThrows(EOFException.class, () -> IOHelper.skipFully(reader, 10)); + } + + /** + * A reader that skips one character per call, which {@link Reader#skip(long)} is free to do and which a single + * unchecked call therefore has to cope with. + */ + private static final class OneAtATimeSkipReader extends FilterReader { + + private OneAtATimeSkipReader(Reader in) { + super(in); + } + + @Override + public long skip(long n) throws IOException { + return n <= 0 ? 0 : super.skip(1); + } + } + } From 9fdf9aea744905e910e0e1668438ae26661cf3c1 Mon Sep 17 00:00:00 2001 From: dhairya05 Date: Thu, 12 Nov 2026 19:12:52 +0530 Subject: [PATCH 2/2] camel-file: close the file when a stale offset runs past its end Review follow-up on #29702. The new EOFException path leaked the open file: in GenericFileConverter.genericFileToInputStream and genericFileToReader the stream is opened and then skipped, and when the skip throws, the stream is neither returned nor closed - so a stale offset leaked a handle on every poll, which on Windows also blocks the move or delete that follows. Both now close it and rethrow. In IOHelper.toString the skip moves inside the try whose finally already closes the reader. Adds the upgrade-guide entry: a stale offset fails the exchange instead of giving an empty body, and the failure repeats until the recorded offset is reset. GenericFileConverterResumeOffsetTest covers both converters skipping to the offset, and an offset past the end of the file failing and releasing the file. --- .../component/file/GenericFileConverter.java | 20 +++- .../GenericFileConverterResumeOffsetTest.java | 102 ++++++++++++++++++ .../java/org/apache/camel/util/IOHelper.java | 4 +- .../pages/camel-4x-upgrade-guide-4_24.adoc | 14 +++ 4 files changed, 134 insertions(+), 6 deletions(-) create mode 100644 components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileConverterResumeOffsetTest.java diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java index ebde0a78dad38..e0e2e3ddc78bd 100644 --- a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java +++ b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConverter.java @@ -124,9 +124,15 @@ public static InputStream genericFileToInputStream(GenericFile file, Exchange LOG.debug("Read file {} (no charset)", f); } InputStream inputStream = IOHelper.toInputStream(f, charset); - // skipNBytes, not skip: a single skip() is allowed to advance fewer bytes than asked for, which - // would leave the body starting in the middle of the data the offset was meant to skip over - inputStream.skipNBytes(file.getLastOffsetValue()); + try { + // skipNBytes, not skip: a single skip() is allowed to advance fewer bytes than asked for, which + // would leave the body starting in the middle of the data the offset was meant to skip over + inputStream.skipNBytes(file.getLastOffsetValue()); + } catch (IOException e) { + // the stream is not returned, so nothing else can close it + IOHelper.close(inputStream); + throw e; + } return inputStream; } @@ -204,7 +210,13 @@ public static Reader genericFileToReader(GenericFile file, Exchange exchange) reader = IOHelper.toReader(f, ExchangeHelper.getCharsetName(exchange)); } - IOHelper.skipFully(reader, file.getLastOffsetValue()); + try { + IOHelper.skipFully(reader, file.getLastOffsetValue()); + } catch (IOException e) { + // the reader is not returned, so nothing else can close it + IOHelper.close(reader); + throw e; + } return reader; } return null; diff --git a/components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileConverterResumeOffsetTest.java b/components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileConverterResumeOffsetTest.java new file mode 100644 index 0000000000000..ebc9b7b34f527 --- /dev/null +++ b/components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileConverterResumeOffsetTest.java @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.file; + +import java.io.EOFException; +import java.io.File; +import java.io.InputStream; +import java.io.Reader; +import java.nio.file.Files; +import java.nio.file.Path; + +import org.apache.camel.util.IOHelper; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The resume support positions the body at a recorded offset. An offset that no longer fits the file - because it was + * truncated or replaced - has to be reported rather than silently yielding a short body, and the file must not be left + * open when it is. + */ +public class GenericFileConverterResumeOffsetTest { + + @TempDir + Path folder; + + @Test + public void shouldSkipToTheOffset() throws Exception { + GenericFile file = fileOf("0123456789", 4); + + try (InputStream is = GenericFileConverter.genericFileToInputStream(file, null)) { + assertEquals("456789", new String(is.readAllBytes())); + } + } + + @Test + public void shouldFailOnAnOffsetPastTheEndOfTheFile() throws Exception { + GenericFile file = fileOf("0123456789", 50); + + assertThrows(EOFException.class, () -> GenericFileConverter.genericFileToInputStream(file, null)); + + // nothing returned the stream, so the converter has to have closed it: on Windows an open handle would + // stop the file being deleted or moved, which is what a failing poll does next + Files.delete(file.getFile().toPath()); + assertTrue(Files.notExists(file.getFile().toPath())); + } + + @Test + public void shouldFailOnAnOffsetPastTheEndOfTheFileForAReader() throws Exception { + GenericFile file = fileOf("0123456789", 50); + file.setCharset("UTF-8"); + + assertThrows(EOFException.class, () -> GenericFileConverter.genericFileToReader(file, null)); + + Files.delete(file.getFile().toPath()); + assertTrue(Files.notExists(file.getFile().toPath())); + } + + @Test + public void shouldSkipToTheOffsetForAReader() throws Exception { + GenericFile file = fileOf("0123456789", 4); + file.setCharset("UTF-8"); + + Reader reader = GenericFileConverter.genericFileToReader(file, null); + try { + assertEquals("456789", IOHelper.toString(reader)); + } finally { + IOHelper.close(reader); + } + } + + private GenericFile fileOf(String content, long offset) throws Exception { + File f = folder.resolve("resume-" + offset + "-" + content.length() + ".txt").toFile(); + Files.writeString(f.toPath(), content); + + GenericFile file = new GenericFile<>(); + file.setFile(f); + file.setFileName(f.getName()); + file.setAbsoluteFilePath(f.getAbsolutePath()); + file.setFileLength(f.length()); + file.updateLastOffsetValue(offset); + return file; + } + +} diff --git a/core/camel-util/src/main/java/org/apache/camel/util/IOHelper.java b/core/camel-util/src/main/java/org/apache/camel/util/IOHelper.java index 76f21379e063b..dcc03906c9fbd 100644 --- a/core/camel-util/src/main/java/org/apache/camel/util/IOHelper.java +++ b/core/camel-util/src/main/java/org/apache/camel/util/IOHelper.java @@ -166,10 +166,10 @@ public static void skipFully(Reader reader, long offset) throws IOException { public static String toString(BufferedReader reader, long offset) throws IOException { StringBuilder sb = new StringBuilder(1024); - skipFully(reader, offset); - char[] buf = new char[1024]; try { + // inside the try, so the reader is closed when a stale offset runs past the end of the content + skipFully(reader, offset); int len; // read until we reach then end which is the -1 marker while ((len = reader.read(buf)) != -1) { diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index d072b4110be32..a7b2133559518 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -8211,3 +8211,17 @@ not pass a language on; a language header alone turns OCR on, unless the endpoin The language itself is passed to docling untouched, in both forms: it is docling's own identifier (language codes such as `en,de`, BCP-47 tags such as `iso:zh-Hant`, or `script/Latin`), and docling validates it. + +=== camel-file - a resume offset past the end of the file now fails the exchange + +The file resume support positions the body at the offset recorded for the file. It did that with a single unchecked +`skip(offset)`, and both `InputStream.skip` and `Reader.skip` are specified to skip *up to* the requested amount, so the +return value is the only thing that says where the stream actually is. It was discarded. + +The reachable case is an offset that no longer fits the file, which is what a recorded offset becomes once the file is +truncated or replaced: the stream ran to the end and the exchange carried an empty or short body as if that were the +file's content. The offset is now honoured in full, and an `EOFException` is thrown when the file ends first. + +If your offsets can go stale, note that the failing exchange leaves the file in place, so the next poll reads the same +offset from the resume store and fails again. Clear or reset the recorded offset for that file - or delete the resume +store entry - to let it be consumed from the start again.