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 7b12fca82cf4 CAMEL-25681: camel-support - FileIdempotentRepository
caches a key only once it is stored
7b12fca82cf4 is described below
commit 7b12fca82cf49980c0d477ffb6370b99c2acef31
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 16 20:27:36 2026 +0530
CAMEL-25681: camel-support - FileIdempotentRepository caches a key only
once it is stored
[CAMEL-25681](https://issues.apache.org/jira/browse/CAMEL-25681)
`FileIdempotentRepository.add(key)` put the key into its 1st-level cache
first, and then checked the file store (`containsStore`) and appended the key
to it (`appendToStore`). When the store could not be read or written (disk
full, permissions, an I/O error of a network file system), `add` threw and the
message failed, but the key stayed in the cache. Every later `add` of that key
returned `false` from the cache until the application was restarted.
So the retry of the failed message was taken for a duplicate and skipped,
also after the store worked again. For example an `idempotentConsumer` with a
file repository behind a JMS consumer: during a "disk full" period every
message fails, the broker redelivers them, and the redeliveries are filtered as
duplicates and acknowledged.
`KafkaIdempotentRepository.add` (camel-kafka, lines 352-363 on main)
already takes the key out of its cache again when storing it fails. The other
repositories in the project have no cache in front of their store.
Closes #27920
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../idempotent/FileIdempotentRepository.java | 9 +-
.../FileIdempotentRepositoryFailedStoreTest.java | 100 +++++++++++++++++++++
2 files changed, 105 insertions(+), 4 deletions(-)
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/processor/idempotent/FileIdempotentRepository.java
b/core/camel-support/src/main/java/org/apache/camel/support/processor/idempotent/FileIdempotentRepository.java
index 81cbfc2a6b6d..b3c9d84a1e99 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/processor/idempotent/FileIdempotentRepository.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/processor/idempotent/FileIdempotentRepository.java
@@ -148,17 +148,18 @@ public class FileIdempotentRepository extends
ServiceSupport implements Idempote
if (cache.containsKey(key)) {
return false;
} else {
- // always register the most used keys in the LRUCache
- cache.put(key, key);
-
// now check the file store
boolean containsInFile = containsStore(key);
if (containsInFile) {
+ // always register the most used keys in the LRUCache
+ cache.put(key, key);
return false;
}
- // its a new key so append to file store
+ // its a new key so append to file store, and only then
register it in the LRUCache: a key that
+ // could not be stored is not added, so that adding it again
is not taken for a duplicate
appendToStore(key);
+ cache.put(key, key);
// check if we hit maximum capacity (if enabled) and report a
warning about this
if (maxFileStoreSize > 0 && fileStore.length() >
maxFileStoreSize) {
diff --git
a/core/camel-support/src/test/java/org/apache/camel/support/processor/idempotent/FileIdempotentRepositoryFailedStoreTest.java
b/core/camel-support/src/test/java/org/apache/camel/support/processor/idempotent/FileIdempotentRepositoryFailedStoreTest.java
new file mode 100644
index 000000000000..dd167f31eeb2
--- /dev/null
+++
b/core/camel-support/src/test/java/org/apache/camel/support/processor/idempotent/FileIdempotentRepositoryFailedStoreTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.support.processor.idempotent;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.RuntimeCamelException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * A key whose add failed because the file store could not be read or written
is not stored, so adding it again (the
+ * retry of the message) must not be taken for a duplicate.
+ */
+class FileIdempotentRepositoryFailedStoreTest {
+
+ @TempDir
+ Path dir;
+
+ private FileIdempotentRepository repository;
+
+ @AfterEach
+ void stop() {
+ if (repository != null) {
+ repository.stop();
+ }
+ }
+
+ @Test
+ void keyIsNotAddedWhenTheStoreCannotBeWritten() throws Exception {
+ AtomicBoolean diskFull = new AtomicBoolean(true);
+ repository = new
FileIdempotentRepository(dir.resolve("store.dat").toFile(), new HashMap<>()) {
+ @Override
+ protected void appendToStore(String key) {
+ if (diskFull.get()) {
+ throw new RuntimeCamelException("No space left on device");
+ }
+ super.appendToStore(key);
+ }
+ };
+ repository.start();
+
+ assertThatThrownBy(() -> repository.add("A")).hasMessageContaining("No
space left on device");
+
+ diskFull.set(false);
+ assertThat(repository.add("A")).as("the retry of a key that was not
stored").isTrue();
+ assertThat(repository.add("A")).as("a duplicate").isFalse();
+
assertThat(Files.readString(dir.resolve("store.dat"))).isEqualTo("A\n");
+ }
+
+ @Test
+ void keyIsNotAddedWhenTheStoreCannotBeRead() throws Exception {
+ File store = dir.resolve("store.dat").toFile();
+ repository = new FileIdempotentRepository(store, new HashMap<>());
+ repository.start();
+
+ // the store cannot be read (here: a directory took its place)
+ assertThat(store.delete()).isTrue();
+ assertThat(store.mkdir()).isTrue();
+ assertThatThrownBy(() ->
repository.add("A")).isInstanceOf(RuntimeCamelException.class);
+
+ assertThat(store.delete()).isTrue();
+ assertThat(repository.add("A")).as("the retry of a key that was not
stored").isTrue();
+ assertThat(repository.contains("A")).isTrue();
+ }
+
+ @Test
+ void keyFoundInTheStoreIsCached() throws Exception {
+ Path store = dir.resolve("store.dat");
+ repository = new FileIdempotentRepository(store.toFile(), new
HashMap<>());
+ repository.start();
+ // written by another repository after this one loaded the store
+ Files.writeString(store, "A\n");
+
+ assertThat(repository.add("A")).isFalse();
+ assertThat(repository.getCache()).containsKey("A");
+ }
+}
