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..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,7 +124,15 @@ public static InputStream genericFileToInputStream(GenericFile file, Exchange LOG.debug("Read file {} (no charset)", f); } InputStream inputStream = IOHelper.toInputStream(f, charset); - inputStream.skip(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; } @@ -202,7 +210,13 @@ public static Reader genericFileToReader(GenericFile file, Exchange exchange) reader = IOHelper.toReader(f, ExchangeHelper.getCharsetName(exchange)); } - reader.skip(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 a660a8a7b02fc..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 @@ -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,13 +141,35 @@ 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); - 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/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); + } + } + } 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.