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 31536afb0374 CAMEL-25505: camel-lucene - Keep inserting after a route
restart, a failed insert or a concurrent insert
31536afb0374 is described below
commit 31536afb0374dd387c68743b919a2630cb9e0b00
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 10 13:06:34 2026 +0530
CAMEL-25505: camel-lucene - Keep inserting after a route restart, a failed
insert or a concurrent insert
A lucene insert endpoint has one LuceneIndexer, with one index directory,
that all its producers share. LuceneIndexProducer.doStop() closed that
directory, so every insert failed after a route restart. An exchange that
failed while it was being indexed left the IndexWriter open with the
index write lock held, so every later insert failed until the JVM
restarted, and two exchanges indexing at the same time failed on that
same lock.
The directory is now closed when the endpoint shuts down instead of when
a producer stops, index() is synchronized, and the writer is rolled back
when adding or committing the documents fails, which releases the lock;
a failing rollback is added as a suppressed exception.
Closes #27638
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
components/camel-lucene/pom.xml | 6 +
.../camel/component/lucene/LuceneEndpoint.java | 10 ++
.../component/lucene/LuceneIndexProducer.java | 6 -
.../camel/component/lucene/LuceneIndexer.java | 32 ++--
.../lucene/LuceneIndexProducerLifecycleTest.java | 193 +++++++++++++++++++++
5 files changed, 231 insertions(+), 16 deletions(-)
diff --git a/components/camel-lucene/pom.xml b/components/camel-lucene/pom.xml
index 3cea1d2468bc..81c277949094 100644
--- a/components/camel-lucene/pom.xml
+++ b/components/camel-lucene/pom.xml
@@ -66,6 +66,12 @@
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility-version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneEndpoint.java
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneEndpoint.java
index 224746a76eb7..7e2debd29201 100644
---
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneEndpoint.java
+++
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneEndpoint.java
@@ -81,6 +81,16 @@ public class LuceneEndpoint extends DefaultEndpoint
implements EndpointServiceLo
return new LuceneIndexProducer(this, this.config, indexer);
}
+ @Override
+ protected void doShutdown() throws Exception {
+ // the index directory is shared by all producers of this endpoint, so
it is closed with the endpoint
+ // and not when a producer (route) stops
+ if (indexer != null) {
+ indexer.getNiofsDirectory().close();
+ }
+ super.doShutdown();
+ }
+
public LuceneConfiguration getConfig() {
return config;
}
diff --git
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneIndexProducer.java
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneIndexProducer.java
index 1410dd1c7f6b..4111cfee2a8f 100644
---
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneIndexProducer.java
+++
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneIndexProducer.java
@@ -30,12 +30,6 @@ public class LuceneIndexProducer extends DefaultProducer {
this.indexer = indexer;
}
- @Override
- public void doStop() throws Exception {
- this.indexer.getNiofsDirectory().close();
- super.doStop();
- }
-
@Override
public void process(Exchange exchange) throws Exception {
indexer.index(exchange);
diff --git
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneIndexer.java
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneIndexer.java
index 3d41fb67a375..4234cb8141d2 100644
---
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneIndexer.java
+++
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneIndexer.java
@@ -70,19 +70,31 @@ public class LuceneIndexer {
}
}
- public void index(Exchange exchange) throws Exception {
+ public synchronized void index(Exchange exchange) throws Exception {
+ // synchronized: the index writer is a field and Lucene allows one
open writer per index directory
LOG.debug("Indexing {}", exchange);
openIndexWriter();
- Map<String, Object> headers = exchange.getIn().getHeaders();
- add("exchangeId", exchange.getExchangeId(), true);
- for (Entry<String, Object> entry : headers.entrySet()) {
- String field = entry.getKey();
- String value =
exchange.getContext().getTypeConverter().mandatoryConvertTo(String.class,
entry.getValue());
- add(field, value, true);
- }
+ try {
+ Map<String, Object> headers = exchange.getIn().getHeaders();
+ add("exchangeId", exchange.getExchangeId(), true);
+ for (Entry<String, Object> entry : headers.entrySet()) {
+ String field = entry.getKey();
+ String value =
exchange.getContext().getTypeConverter().mandatoryConvertTo(String.class,
entry.getValue());
+ add(field, value, true);
+ }
- add("contents", exchange.getIn().getMandatoryBody(String.class), true);
- closeIndexWriter();
+ add("contents", exchange.getIn().getMandatoryBody(String.class),
true);
+ indexWriter.commit();
+ } catch (Exception e) {
+ // discard the documents of this exchange and release the index
write lock
+ try {
+ indexWriter.rollback();
+ } catch (Exception re) {
+ e.addSuppressed(re);
+ }
+ throw e;
+ }
+ indexWriter.close();
}
public NIOFSDirectory getNiofsDirectory() {
diff --git
a/components/camel-lucene/src/test/java/org/apache/camel/component/lucene/LuceneIndexProducerLifecycleTest.java
b/components/camel-lucene/src/test/java/org/apache/camel/component/lucene/LuceneIndexProducerLifecycleTest.java
new file mode 100644
index 000000000000..5ef5a1991651
--- /dev/null
+++
b/components/camel-lucene/src/test/java/org/apache/camel/component/lucene/LuceneIndexProducerLifecycleTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.lucene;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.lucene.analysis.standard.StandardAnalyzer;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.StoredFields;
+import org.apache.lucene.store.FSDirectory;
+import org.apache.lucene.store.NIOFSDirectory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The insert endpoint owns one index directory that all its producers share:
stopping a route must not close it, and
+ * concurrent exchanges must not fail on the index write lock.
+ */
+class LuceneIndexProducerLifecycleTest extends CamelTestSupport {
+
+ private static final String INSERT =
"lucene:lifecycleIndex:insert?analyzer=#stdAnalyzer&indexDir=#lifecycleDir";
+
+ @TempDir
+ File indexDir;
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("lifecycleDir", indexDir);
+ registry.bind("stdAnalyzer", new StandardAnalyzer());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:index").routeId("index").to(INSERT);
+ from("direct:other").routeId("other").to(INSERT);
+ }
+ };
+ }
+
+ @Test
+ void insertWorksAfterRouteRestart() throws Exception {
+ Exchange first = template.send("direct:index", e ->
e.getIn().setBody("before restart"));
+ assertNull(first.getException(), "insert before the restart");
+
+ context.getRouteController().stopRoute("index");
+ context.getRouteController().startRoute("index");
+
+ Exchange second = template.send("direct:index", e ->
e.getIn().setBody("after restart"));
+ assertNull(second.getException(), "an insert after the route was
restarted must succeed");
+ assertEquals(2, countIndexedMessages());
+ }
+
+ @Test
+ void stoppingOneRouteDoesNotBreakAnotherRouteOnTheSameEndpoint() throws
Exception {
+ context.getRouteController().stopRoute("other");
+
+ Exchange exchange = template.send("direct:index", e ->
e.getIn().setBody("still running"));
+ assertNull(exchange.getException(),
+ "stopping route 'other' must not break the insert of route
'index' to the same endpoint");
+ assertEquals(1, countIndexedMessages());
+ }
+
+ @Test
+ void failedInsertDoesNotBlockLaterInserts() throws Exception {
+ // no body: the insert fails after the index writer was opened
+ Exchange failed = template.send("direct:index", e ->
e.getIn().setBody(null));
+ assertNotNull(failed.getException(), "an insert without a body fails");
+
+ Exchange next = template.send("direct:index", e ->
e.getIn().setBody("next"));
+ assertNull(next.getException(), "a failed insert must not leave the
index write lock held");
+ assertEquals(1, countIndexedMessages());
+ }
+
+ @Test
+ void failedCommitDoesNotBlockLaterInserts() throws Exception {
+ LuceneEndpoint endpoint = context.getEndpoint(INSERT,
LuceneEndpoint.class);
+ LuceneIndexer indexer = ((LuceneIndexProducer)
endpoint.createProducer()).getIndexer();
+ AtomicBoolean failSync = new AtomicBoolean(true);
+ NIOFSDirectory original = indexer.getNiofsDirectory();
+ // the first commit fails when it syncs the index files
+ indexer.setNiofsDirectory(new NIOFSDirectory(indexDir.toPath()) {
+ @Override
+ public void sync(Collection<String> names) throws IOException {
+ if (failSync.getAndSet(false)) {
+ throw new IOException("Simulated commit failure");
+ }
+ super.sync(names);
+ }
+ });
+ original.close();
+
+ Exchange failed = template.send("direct:index", e ->
e.getIn().setBody("commit fails"));
+ assertNotNull(failed.getException(), "an insert whose commit fails
fails");
+ assertFalse(failSync.get(), "the commit should have synced the index
files");
+
+ Exchange next = template.send("direct:index", e ->
e.getIn().setBody("next"));
+ assertNull(next.getException(), "a failed commit must not leave the
index write lock held");
+ assertEquals(1, countIndexedMessages());
+ }
+
+ @Test
+ void concurrentInsertsAreIndexed() throws Exception {
+ CountDownLatch writerOpen = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ // a header value whose conversion to String happens while the first
insert holds the index writer
+ Object slowHeader = new Object() {
+ @Override
+ public String toString() {
+ writerOpen.countDown();
+ try {
+ release.await(20, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return "slow";
+ }
+ };
+
+ AtomicReference<Exchange> firstResult = new AtomicReference<>();
+ AtomicReference<Exchange> secondResult = new AtomicReference<>();
+ Thread first = new Thread(() ->
firstResult.set(template.send("direct:index", e -> {
+ e.getIn().setHeader("slow", slowHeader);
+ e.getIn().setBody("first");
+ })));
+ first.start();
+ assertTrue(writerOpen.await(10, TimeUnit.SECONDS), "first insert
should hold the index writer");
+
+ Thread second = new Thread(() ->
secondResult.set(template.send("direct:other", e ->
e.getIn().setBody("second"))));
+ second.start();
+ // the second insert either completes or waits for the first one
+ await().atMost(10, TimeUnit.SECONDS).until(() -> !second.isAlive()
+ || second.getState() == Thread.State.BLOCKED ||
second.getState() == Thread.State.WAITING);
+
+ release.countDown();
+ first.join(10000);
+ second.join(10000);
+ assertFalse(first.isAlive(), "first insert should have completed");
+ assertFalse(second.isAlive(), "second insert should have completed");
+
+ assertNull(firstResult.get().getException(), "first insert");
+ assertNull(secondResult.get().getException(),
+ "an insert running while another insert to the same index is
in progress must not fail");
+ assertEquals(2, countIndexedMessages());
+ }
+
+ private int countIndexedMessages() throws IOException {
+ int count = 0;
+ try (FSDirectory dir = FSDirectory.open(indexDir.toPath());
+ DirectoryReader reader = DirectoryReader.open(dir)) {
+ StoredFields storedFields = reader.storedFields();
+ for (int i = 0; i < reader.maxDoc(); i++) {
+ if (storedFields.document(i).get("contents") != null) {
+ count++;
+ }
+ }
+ }
+ return count;
+ }
+}
