This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 18755f062c98 camel-file: do not assume skip() advanced the whole
offset (#29702)
18755f062c98 is described below
commit 18755f062c989b74e40eda98709efa902be9a185
Author: dhairya05 <[email protected]>
AuthorDate: Sun Nov 15 01:06:13 2026 +0530
camel-file: do not assume skip() advanced the whole offset (#29702)
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.
Co-authored-by: Guillaume Nodet - AI Bot <[email protected]>
---
.../camel/component/file/GenericFileConverter.java | 18 +++-
.../file/GenericFileConverterResumeOffsetTest.java | 102 +++++++++++++++++++++
.../main/java/org/apache/camel/util/IOHelper.java | 27 +++++-
.../java/org/apache/camel/util/IOHelperTest.java | 36 ++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 +++
5 files changed, 193 insertions(+), 4 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 2c7dd3933f9b..e0e2e3ddc78b 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 final class GenericFileConverter {
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 final class GenericFileConverter {
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 000000000000..ebc9b7b34f52
--- /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> 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> 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> 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> 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<File> fileOf(String content, long offset) throws
Exception {
+ File f = folder.resolve("resume-" + offset + "-" + content.length() +
".txt").toFile();
+ Files.writeString(f.toPath(), content);
+
+ GenericFile<File> 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 a660a8a7b02f..dcc03906c9fb 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.BufferedReader;
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 final class IOHelper {
return toString(reader, INITIAL_OFFSET);
}
+ /**
+ * Skips exactly {@code offset} characters of the reader.
+ * <p/>
+ * {@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 8e652e88753b..153bcc08e64b 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 @@ package org.apache.camel.util;
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 class IOHelperTest {
}
}
+ @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 33a70db846d5..105b6ed66006 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
@@ -8825,3 +8825,17 @@ if it configured the proxy.
It is now deprecated, with a `deprecationNote` that says so, and its
description no longer claims to configure a proxy.
Remove `protocol` from the endpoint URI or the configuration. To send the
client's traffic through a proxy, use
`proxyProtocol` together with `proxyHost` and `proxyPort`, which the shared
client builder does apply.
+
+=== 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 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 by default (without `moveFailed`) 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.
