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 e78477968620 CAMEL-25506: camel-lucene - Query with a searcher per
exchange and do not close the analyzer on stop
e78477968620 is described below
commit e7847796862013ab78d278d20ee96ef2af21b0bd
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 10 13:06:45 2026 +0530
CAMEL-25506: camel-lucene - Query with a searcher per exchange and do not
close the analyzer on stop
LuceneQueryProducer used one LuceneSearcher for all exchanges, and
LuceneSearcher kept the reader, searcher and hits of the last query in
fields, so concurrent exchanges could get the hits of another exchange's
query and the readers of earlier queries were never closed. doStop()
closed the endpoint's analyzer, so every query failed after a route
restart.
LuceneQueryProducer and LuceneQueryProcessor now open, search and close a
LuceneSearcher per exchange, and LuceneSearcher.close() closes only its
reader, not the analyzer, which belongs to the endpoint or the registry.
Code that uses LuceneSearcher directly must now close its analyzer
itself, as noted in the 4.23 upgrade guide.
Closes #27639
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/lucene/LuceneQueryProducer.java | 21 +--
.../camel/component/lucene/LuceneSearcher.java | 13 +-
.../processor/lucene/LuceneQueryProcessor.java | 10 +-
.../lucene/LuceneQueryProducerLifecycleTest.java | 151 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 5 +
5 files changed, 181 insertions(+), 19 deletions(-)
diff --git
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneQueryProducer.java
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneQueryProducer.java
index 9da9bea1758d..cc97919fb5ef 100644
---
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneQueryProducer.java
+++
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneQueryProducer.java
@@ -26,7 +26,6 @@ import org.apache.lucene.analysis.Analyzer;
public class LuceneQueryProducer extends DefaultProducer {
LuceneConfiguration config;
- LuceneSearcher searcher;
Analyzer analyzer;
File indexDirectory;
int maxNumberOfHits;
@@ -40,18 +39,6 @@ public class LuceneQueryProducer extends DefaultProducer {
maxNumberOfHits = config.getMaxHits();
}
- @Override
- public void doStart() throws Exception {
- searcher = new LuceneSearcher();
- super.doStart();
- }
-
- @Override
- public void doStop() throws Exception {
- searcher.close();
- super.doStop();
- }
-
@Override
public void process(Exchange exchange) throws Exception {
Hits hits;
@@ -61,8 +48,14 @@ public class LuceneQueryProducer extends DefaultProducer {
boolean isReturnLuceneDocs = returnLuceneDocs != null &&
returnLuceneDocs.equalsIgnoreCase("true");
if (phrase != null) {
+ // a searcher (and index reader) per query: the producer is shared
by concurrent exchanges
+ LuceneSearcher searcher = new LuceneSearcher();
searcher.open(indexDirectory, analyzer);
- hits = searcher.search(phrase, maxNumberOfHits,
totalHitsThreshold, isReturnLuceneDocs);
+ try {
+ hits = searcher.search(phrase, maxNumberOfHits,
totalHitsThreshold, isReturnLuceneDocs);
+ } finally {
+ searcher.close();
+ }
} else {
throw new IllegalArgumentException(
"SearchPhrase for LucenePhraseQuerySearcher not set. Set
the Header value: QUERY");
diff --git
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneSearcher.java
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneSearcher.java
index a354379edfe1..40f5481c7c60 100644
---
a/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneSearcher.java
+++
b/components/camel-lucene/src/main/java/org/apache/camel/component/lucene/LuceneSearcher.java
@@ -35,6 +35,9 @@ import org.apache.lucene.store.NIOFSDirectory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Runs one query on its own reader of the index. An instance is used by one
exchange at a time: open, search, close.
+ */
public class LuceneSearcher {
private static final Logger LOG =
LoggerFactory.getLogger(LuceneSearcher.class);
private Analyzer analyzer;
@@ -52,9 +55,15 @@ public class LuceneSearcher {
this.analyzer = analyzer;
}
+ /**
+ * Closes the index reader. The analyzer is not closed: it belongs to the
endpoint (or the registry) and is used by
+ * the next query.
+ */
public void close() throws IOException {
- indexReader.close();
- analyzer.close();
+ if (indexReader != null) {
+ indexReader.close();
+ indexReader = null;
+ }
}
public Hits search(String searchPhrase, int maxNumberOfHits, int
totalHitsThreshold) throws Exception {
diff --git
a/components/camel-lucene/src/main/java/org/apache/camel/processor/lucene/LuceneQueryProcessor.java
b/components/camel-lucene/src/main/java/org/apache/camel/processor/lucene/LuceneQueryProcessor.java
index 00e5e023e96b..cbf0a0f53d1d 100644
---
a/components/camel-lucene/src/main/java/org/apache/camel/processor/lucene/LuceneQueryProcessor.java
+++
b/components/camel-lucene/src/main/java/org/apache/camel/processor/lucene/LuceneQueryProcessor.java
@@ -30,7 +30,6 @@ public class LuceneQueryProcessor implements Processor {
private File indexDirectory;
private Analyzer analyzer;
private IndexSearcher indexSearcher;
- private LuceneSearcher searcher;
private String searchPhrase;
private int maxNumberOfHits;
private int totalHitsThreshold;
@@ -51,9 +50,14 @@ public class LuceneQueryProcessor implements Processor {
boolean isReturnLuceneDocs = returnLuceneDocs != null &&
returnLuceneDocs.equalsIgnoreCase("true");
if (phrase != null) {
- searcher = new LuceneSearcher();
+ // a searcher (and index reader) per query, closed when the hits
are built
+ LuceneSearcher searcher = new LuceneSearcher();
searcher.open(indexDirectory, analyzer);
- hits = searcher.search(phrase, maxNumberOfHits,
totalHitsThreshold, isReturnLuceneDocs);
+ try {
+ hits = searcher.search(phrase, maxNumberOfHits,
totalHitsThreshold, isReturnLuceneDocs);
+ } finally {
+ searcher.close();
+ }
} else {
throw new IllegalArgumentException("SearchPhrase for
LuceneQueryProcessor not set. Set the Header value: QUERY");
}
diff --git
a/components/camel-lucene/src/test/java/org/apache/camel/component/lucene/LuceneQueryProducerLifecycleTest.java
b/components/camel-lucene/src/test/java/org/apache/camel/component/lucene/LuceneQueryProducerLifecycleTest.java
new file mode 100644
index 000000000000..80ee31c6e7ff
--- /dev/null
+++
b/components/camel-lucene/src/test/java/org/apache/camel/component/lucene/LuceneQueryProducerLifecycleTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.lucene.support.Hit;
+import org.apache.camel.processor.lucene.support.Hits;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.lucene.analysis.standard.StandardAnalyzer;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.store.FSDirectory;
+import org.junit.jupiter.api.BeforeEach;
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The query producer must keep working after its route is restarted, must
stop cleanly when it never queried, and must
+ * answer each exchange with the hits of its own query.
+ */
+class LuceneQueryProducerLifecycleTest extends CamelTestSupport {
+
+ @TempDir
+ static File indexDir;
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("queryDir", indexDir);
+ registry.bind("queryAnalyzer", new StandardAnalyzer());
+ }
+
+ @BeforeEach
+ void createIndex() throws Exception {
+ try (FSDirectory dir = FSDirectory.open(indexDir.toPath());
+ IndexWriter writer = new IndexWriter(
+ dir, new IndexWriterConfig(new
StandardAnalyzer()).setOpenMode(IndexWriterConfig.OpenMode.CREATE))) {
+ for (int i = 0; i < 20; i++) {
+ addDoc(writer, "alpha document " + i);
+ addDoc(writer, "beta document " + i);
+ }
+ writer.commit();
+ }
+ }
+
+ private static void addDoc(IndexWriter writer, String contents) throws
Exception {
+ Document doc = new Document();
+ doc.add(new TextField("contents", contents, Field.Store.YES));
+ writer.addDocument(doc);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:query").routeId("query")
+
.to("lucene:queryIndex:query?analyzer=#queryAnalyzer&indexDir=#queryDir&maxHits=20");
+ }
+ };
+ }
+
+ private Exchange query(String phrase) {
+ return template.send("direct:query", e ->
e.getIn().setHeader(LuceneConstants.HEADER_QUERY, phrase));
+ }
+
+ @Test
+ void queryWorksAfterRouteRestart() throws Exception {
+ Exchange before = query("alpha");
+ assertNull(before.getException(), "query before the restart");
+ assertEquals(20,
before.getMessage().getBody(Hits.class).getNumberOfHits());
+
+ context.getRouteController().stopRoute("query");
+ context.getRouteController().startRoute("query");
+
+ Exchange after = query("alpha");
+ assertNull(after.getException(), "a query after the route was
restarted must succeed");
+ assertEquals(20,
after.getMessage().getBody(Hits.class).getNumberOfHits());
+ }
+
+ /**
+ * The producer is shared by concurrent exchanges. Before the fix, its
searcher kept the reader, the searcher and
+ * the hits of the last query in fields, so an exchange could get the hits
of another exchange's query (a few dozen
+ * of 20000 queries on a laptop). This test cannot force that
interleaving, it only makes it very likely.
+ */
+ @Test
+ void concurrentQueriesGetTheirOwnHits() throws Exception {
+ ExecutorService pool = Executors.newFixedThreadPool(8);
+ try {
+ List<Future<String>> results = new ArrayList<>();
+ for (int i = 0; i < 20000; i++) {
+ String term = i % 2 == 0 ? "alpha" : "beta";
+ Callable<String> task = () -> {
+ Exchange exchange = query(term);
+ if (exchange.getException() != null) {
+ return term + ": " + exchange.getException();
+ }
+ for (Hit hit :
exchange.getMessage().getBody(Hits.class).getHit()) {
+ if (!hit.getData().startsWith(term)) {
+ return term + ": got hit '" + hit.getData() + "'";
+ }
+ }
+ return null;
+ };
+ results.add(pool.submit(task));
+ }
+ List<String> wrong = new ArrayList<>();
+ for (Future<String> result : results) {
+ String error = result.get(30, TimeUnit.SECONDS);
+ if (error != null) {
+ wrong.add(error);
+ }
+ }
+ assertTrue(wrong.isEmpty(), "each query must get the hits of its
own query, but " + wrong.size()
+ + " of 20000 did not, e.g. " +
wrong.stream().limit(3).toList());
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 7b50fb025307..4cfbd7d495c8 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -2593,6 +2593,11 @@ would pass `null` to the underlying store's `remove()`
method, which threw an op
from langchain4j. It now throws `IllegalArgumentException` with a clear
message listing the
expected inputs.
+=== camel-lucene
+
+`LuceneSearcher.close()` no longer closes the `Analyzer` passed to `open()`;
code that uses `LuceneSearcher` directly
+must close its analyzer itself when it is no longer needed.
+
=== camel-ftp, camel-sftp, camel-ftps, camel-mina-sftp, camel-azure-files,
camel-smb
The remote-file consumers now ensure the path resolved for a polled file stays
within the directory being
