You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stormcrawler.apache.org by rz...@apache.org on 2026/07/07 18:06:34 UTC
(stormcrawler) 01/01: Port QueueBolt (queue stream) to the opensearch-java module
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch port-queuebolt-opensearch-java
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit a19f72edcb9ce63ba91838e7cf2fa7541f3b08da
Author: Richard Zowalla <rz...@apache.org>
AuthorDate: Tue Jul 7 20:06:20 2026 +0200
Port QueueBolt (queue stream) to the opensearch-java module
Ports the queue stream feature (#1974) from external/opensearch to the
newer external/opensearch-java module, which uses the opensearch-java
client instead of the legacy REST high-level client.
---
.../opensearch/persistence/QueueBolt.java | 191 +++++++++++++++++++++
.../opensearch/persistence/StatusUpdaterBolt.java | 7 +
.../{StatusBoltTest.java => QueueBoltTest.java} | 116 +++++++++----
.../opensearch/bolt/StatusBoltTest.java | 22 +++
.../src/test/resources/queues.mapping | 33 ++++
5 files changed, 335 insertions(+), 34 deletions(-)
diff --git a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/QueueBolt.java b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/QueueBolt.java
new file mode 100644
index 00000000..6af7d909
--- /dev/null
+++ b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/QueueBolt.java
@@ -0,0 +1,191 @@
+/*
+ * 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.stormcrawler.opensearch.persistence;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.tuple.Tuple;
+import org.apache.stormcrawler.opensearch.AsyncBulkProcessor;
+import org.apache.stormcrawler.opensearch.Constants;
+import org.apache.stormcrawler.opensearch.IndexCreation;
+import org.apache.stormcrawler.opensearch.OpenSearchConnection;
+import org.apache.stormcrawler.util.ConfUtils;
+import org.opensearch.client.opensearch.core.BulkRequest;
+import org.opensearch.client.opensearch.core.BulkResponse;
+import org.opensearch.client.opensearch.core.bulk.BulkOperation;
+import org.opensearch.client.opensearch.core.bulk.BulkResponseItem;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Sends information about the queues into an OpenSearch index. This has to be connected to a status
+ * updater bolt.
+ */
+public class QueueBolt extends BaseRichBolt implements AsyncBulkProcessor.Listener {
+
+ private static final Logger LOG = LoggerFactory.getLogger(QueueBolt.class);
+
+ private static final String OSBoltType = "queues";
+
+ static final String OSQueuesIndexNameParamName =
+ Constants.PARAMPREFIX + OSBoltType + ".index.name";
+
+ /**
+ * Parameter name to configure the cache of known queues. @see
+ * https://github.com/ben-manes/caffeine/wiki/Specification Default value is
+ * "maximumSize=10000".
+ */
+ static final String OSQueuesCacheSpecParamName =
+ Constants.PARAMPREFIX + OSBoltType + ".cache.spec";
+
+ private OutputCollector _collector;
+
+ private String indexName;
+
+ private OpenSearchConnection connection;
+
+ /** Keys confirmed to be present in the index; populated on bulk success only */
+ private Cache<String, Boolean> knownQueue;
+
+ /** docID to key for the requests currently in the bulk processor */
+ private final Map<String, String> inFlight = new ConcurrentHashMap<>();
+
+ public QueueBolt() {}
+
+ /** Sets the index name instead of taking it from the configuration. * */
+ public QueueBolt(String indexName) {
+ this.indexName = indexName;
+ }
+
+ @Override
+ public void prepare(
+ Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
+ _collector = collector;
+ if (indexName == null) {
+ indexName = ConfUtils.getString(conf, QueueBolt.OSQueuesIndexNameParamName, OSBoltType);
+ }
+ try {
+ connection = OpenSearchConnection.getConnection(conf, OSBoltType, this);
+ } catch (Exception e1) {
+ LOG.error("Can't connect to OpenSearch", e1);
+ throw new RuntimeException(e1);
+ }
+ try {
+ IndexCreation.checkOrCreateIndex(connection.getClient(), indexName, OSBoltType, LOG);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ final String cacheSpec =
+ ConfUtils.getString(
+ conf, QueueBolt.OSQueuesCacheSpecParamName, "maximumSize=10000");
+ knownQueue = Caffeine.from(cacheSpec).build();
+ }
+
+ @Override
+ public void cleanup() {
+ if (connection != null) {
+ connection.close();
+ }
+ }
+
+ @Override
+ public void execute(Tuple tuple) {
+
+ final String key = tuple.getStringByField("key");
+
+ // ack no matter what - the queue info is best effort bookkeeping
+ // and failed writes are retried when the key is next seen
+ _collector.ack(tuple);
+
+ // check whether this key is already known
+ if (knownQueue.getIfPresent(key) != null) {
+ return;
+ }
+
+ final String docID = org.apache.commons.codec.digest.DigestUtils.sha256Hex(key);
+
+ // a request for this key is already in the bulk processor
+ if (inFlight.putIfAbsent(docID, key) != null) {
+ return;
+ }
+
+ final HashMap<String, String> fields = new HashMap<>();
+ fields.put("key", key);
+ fields.put("lastUpdated", Instant.now().toString());
+
+ final BulkOperation op =
+ BulkOperation.of(b -> b.create(c -> c.index(indexName).id(docID).document(fields)));
+
+ connection.addToProcessor(op);
+ }
+
+ @Override
+ public void beforeBulk(long executionId, BulkRequest request) {
+ LOG.debug("beforeBulk {} with {} actions", executionId, request.operations().size());
+ }
+
+ @Override
+ public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
+ LOG.debug("afterBulk [{}] with {} responses", executionId, response.items().size());
+ for (BulkResponseItem item : response.items()) {
+ final String key = inFlight.remove(item.id());
+ if (key == null) {
+ continue;
+ }
+ if (item.error() == null) {
+ knownQueue.put(key, Boolean.TRUE);
+ } else if (item.status() == 409) {
+ // entry already exists in the index, e.g. written before a worker
+ // restart or evicted from the cache - not an error
+ knownQueue.put(key, Boolean.TRUE);
+ } else {
+ // do not cache the key so that the write is retried
+ // when the key is next seen
+ final var error = item.error();
+ LOG.error(
+ "Failed to index queue entry {}: {}",
+ key,
+ error.reason() != null ? error.reason() : error.type());
+ }
+ }
+ }
+
+ @Override
+ public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
+ LOG.error("Exception with bulk {} - failing the whole lot ", executionId, failure);
+ for (BulkOperation op : request.operations()) {
+ // not cached - the writes are retried when the keys are next seen
+ inFlight.remove(OpenSearchConnection.getBulkOperationId(op));
+ }
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ // nothing to do here - this bolt is the last of a topology
+ }
+}
diff --git a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java
index 93626a92..bc0c59cc 100644
--- a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java
+++ b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java
@@ -28,6 +28,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
import org.apache.stormcrawler.Metadata;
import org.apache.stormcrawler.metrics.CrawlerMetrics;
import org.apache.stormcrawler.metrics.ScopedCounter;
@@ -238,6 +239,12 @@ public class StatusUpdaterBolt extends AbstractStatusUpdaterBolt
partitionKey = "_DEFAULT_";
}
+ // send a tuple on the queue stream in case a bolt
+ // wants to handle it
+ super.collector.emit(
+ org.apache.stormcrawler.Constants.QUEUE_STREAM_NAME,
+ new Values(partitionKey, metadata));
+
// store routing key in metadata?
if (StringUtils.isNotBlank(fieldNameForRoutingKey) && routingFieldNameInMetadata) {
metadataMap.put(fieldNameForRoutingKey, partitionKey);
diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/StatusBoltTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/QueueBoltTest.java
similarity index 56%
copy from external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/StatusBoltTest.java
copy to external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/QueueBoltTest.java
index 929ae5c1..296c58dd 100644
--- a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/StatusBoltTest.java
+++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/QueueBoltTest.java
@@ -19,7 +19,6 @@ package org.apache.stormcrawler.opensearch.bolt;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -38,8 +37,7 @@ import org.apache.storm.tuple.Tuple;
import org.apache.stormcrawler.Metadata;
import org.apache.stormcrawler.TestOutputCollector;
import org.apache.stormcrawler.TestUtil;
-import org.apache.stormcrawler.opensearch.persistence.StatusUpdaterBolt;
-import org.apache.stormcrawler.persistence.Status;
+import org.apache.stormcrawler.opensearch.persistence.QueueBolt;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
@@ -54,9 +52,9 @@ import org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportBui
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-class StatusBoltTest extends AbstractOpenSearchTest {
+class QueueBoltTest extends AbstractOpenSearchTest {
- private StatusUpdaterBolt bolt;
+ private QueueBolt bolt;
protected TestOutputCollector output;
@@ -64,7 +62,7 @@ class StatusBoltTest extends AbstractOpenSearchTest {
private OpenSearchTransport transport;
- private static final Logger LOG = LoggerFactory.getLogger(StatusBoltTest.class);
+ private static final Logger LOG = LoggerFactory.getLogger(QueueBoltTest.class);
private static ExecutorService executorService;
@@ -80,8 +78,7 @@ class StatusBoltTest extends AbstractOpenSearchTest {
}
@BeforeEach
- void setupStatusBolt() throws IOException {
- bolt = new StatusUpdaterBolt();
+ void setupQueueBolt() throws IOException {
transport =
ApacheHttpClient5TransportBuilder.builder(
new HttpHost(
@@ -91,22 +88,23 @@ class StatusBoltTest extends AbstractOpenSearchTest {
.setMapper(new JacksonJsonpMapper())
.build();
client = new OpenSearchClient(transport);
- // configure the status updater bolt
+ output = new TestOutputCollector();
+ bolt = prepareQueueBolt();
+ }
+
+ private QueueBolt prepareQueueBolt() {
+ QueueBolt queueBolt = new QueueBolt();
Map<String, Object> conf = new HashMap<>();
- conf.put("opensearch.status.routing.fieldname", "metadata.key");
conf.put(
- "opensearch.status.addresses",
+ "opensearch.queues.addresses",
opensearchContainer.getHost() + ":" + opensearchContainer.getFirstMappedPort());
- conf.put("scheduler.class", "org.apache.stormcrawler.persistence.DefaultScheduler");
- conf.put("status.updater.cache.spec", "maximumSize=10000,expireAfterAccess=1h");
- conf.put("metadata.persist", "someKey");
- output = new TestOutputCollector();
- bolt.prepare(conf, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+ queueBolt.prepare(conf, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+ return queueBolt;
}
@AfterEach
void close() {
- LOG.info("Closing updater bolt and Opensearch container");
+ LOG.info("Closing queue bolt and Opensearch container");
super.close();
bolt.cleanup();
output = null;
@@ -116,10 +114,9 @@ class StatusBoltTest extends AbstractOpenSearchTest {
}
}
- private Future<Integer> store(String url, Status status, Metadata metadata) {
+ private Future<Integer> store(String key, Metadata metadata) {
Tuple tuple = mock(Tuple.class);
- when(tuple.getValueByField("status")).thenReturn(status);
- when(tuple.getStringByField("url")).thenReturn(url);
+ when(tuple.getStringByField("key")).thenReturn(key);
when(tuple.getValueByField("metadata")).thenReturn(metadata);
bolt.execute(tuple);
return executorService.submit(
@@ -130,25 +127,76 @@ class StatusBoltTest extends AbstractOpenSearchTest {
});
}
+ private void awaitIndexed(String key) {
+ String id = org.apache.commons.codec.digest.DigestUtils.sha256Hex(key);
+ await().atMost(30, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ try {
+ GetResponse<Map> result =
+ client.get(g -> g.index("queues").id(id), Map.class);
+ return result.found();
+ } catch (IOException e) {
+ return false;
+ }
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<String, Object> getSource(String key) throws IOException {
+ String id = org.apache.commons.codec.digest.DigestUtils.sha256Hex(key);
+ return client.get(g -> g.index("queues").id(id), Map.class).source();
+ }
+
@Test
@Timeout(value = 2, unit = TimeUnit.MINUTES)
- // see https://github.com/apache/stormcrawler/issues/885
- @SuppressWarnings("unchecked")
- void checkListKeyFromOpensearch()
+ void checkQueueIndex()
throws IOException, ExecutionException, InterruptedException, TimeoutException {
- String url = "https://www.url.net/something";
+ String key = "www.url.net";
Metadata md = new Metadata();
md.addValue("someKey", "someValue");
- store(url, Status.DISCOVERED, md).get(10, TimeUnit.SECONDS);
+ store(key, md).get(10, TimeUnit.SECONDS);
assertEquals(1, output.getAckedTuples().size());
- // check output in Opensearch
- String id = org.apache.commons.codec.digest.DigestUtils.sha256Hex(url);
- GetResponse<Map> result = client.get(g -> g.index("status").id(id), Map.class);
- Map<String, Object> sourceAsMap = result.source();
- final String pfield = "metadata.somekey";
- sourceAsMap = (Map<String, Object>) sourceAsMap.get("metadata");
- final var pfieldNew = pfield.substring(9);
- Object key = sourceAsMap.get(pfieldNew);
- assertTrue(key instanceof java.util.ArrayList);
+
+ // Wait until document is indexed in OpenSearch
+ awaitIndexed(key);
+
+ Map<String, Object> sourceAsMap = getSource(key);
+ assertEquals(key, sourceAsMap.get("key"));
+ }
+
+ @Test
+ @Timeout(value = 2, unit = TimeUnit.MINUTES)
+ void checkExistingEntryIsNotOverwritten()
+ throws IOException, ExecutionException, InterruptedException, TimeoutException {
+ String key = "www.url.net";
+ Metadata md = new Metadata();
+ md.addValue("someKey", "someValue");
+ store(key, md).get(10, TimeUnit.SECONDS);
+ awaitIndexed(key);
+ Object lastUpdated = getSource(key).get("lastUpdated");
+
+ // a fresh bolt instance (e.g. after a worker restart) has an empty cache
+ // and re-sends the same key - the resulting version conflict must be
+ // handled gracefully and the existing entry left untouched
+ QueueBolt restartedBolt = prepareQueueBolt();
+ try {
+ Tuple sameKey = mock(Tuple.class);
+ when(sameKey.getStringByField("key")).thenReturn(key);
+ restartedBolt.execute(sameKey);
+
+ // a new key sent in the same bulk signals when the bulk has been processed
+ String otherKey = "www.other.net";
+ Tuple other = mock(Tuple.class);
+ when(other.getStringByField("key")).thenReturn(otherKey);
+ restartedBolt.execute(other);
+ awaitIndexed(otherKey);
+
+ Map<String, Object> sourceAsMap = getSource(key);
+ assertEquals(key, sourceAsMap.get("key"));
+ assertEquals(lastUpdated, sourceAsMap.get("lastUpdated"));
+ } finally {
+ restartedBolt.cleanup();
+ }
}
}
diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/StatusBoltTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/StatusBoltTest.java
index 929ae5c1..71fe7795 100644
--- a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/StatusBoltTest.java
+++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/bolt/StatusBoltTest.java
@@ -151,4 +151,26 @@ class StatusBoltTest extends AbstractOpenSearchTest {
Object key = sourceAsMap.get(pfieldNew);
assertTrue(key instanceof java.util.ArrayList);
}
+
+ @Test
+ @Timeout(value = 2, unit = TimeUnit.MINUTES)
+ void checkQueueStreamEmission()
+ throws IOException, ExecutionException, InterruptedException, TimeoutException {
+ String url = "https://www.url.net/something";
+ Metadata md = new Metadata();
+ md.addValue("someKey", "someValue");
+ store(url, Status.DISCOVERED, md).get(10, TimeUnit.SECONDS);
+ assertEquals(1, output.getAckedTuples().size());
+
+ // Check that a tuple was emitted on the queue stream
+ var emittedQueue = output.getEmitted(org.apache.stormcrawler.Constants.QUEUE_STREAM_NAME);
+ assertEquals(1, emittedQueue.size());
+
+ var queueTuple = emittedQueue.get(0);
+ assertEquals(2, queueTuple.size());
+ assertEquals("www.url.net", queueTuple.get(0));
+ assertTrue(queueTuple.get(1) instanceof Metadata);
+ Metadata emittedMetadata = (Metadata) queueTuple.get(1);
+ assertEquals("someValue", emittedMetadata.getFirstValue("someKey"));
+ }
}
diff --git a/external/opensearch-java/src/test/resources/queues.mapping b/external/opensearch-java/src/test/resources/queues.mapping
new file mode 100644
index 00000000..b85f33dc
--- /dev/null
+++ b/external/opensearch-java/src/test/resources/queues.mapping
@@ -0,0 +1,33 @@
+{
+ "settings": {
+ "index": {
+ "number_of_shards": 10,
+ "number_of_replicas": 1,
+ "refresh_interval": "5s"
+ }
+ },
+ "mappings": {
+ "dynamic_templates": [{
+ "metadata": {
+ "path_match": "metadata.*",
+ "match_mapping_type": "string",
+ "mapping": {
+ "type": "keyword"
+ }
+ }
+ }],
+ "_source": {
+ "enabled": true
+ },
+ "properties": {
+ "key": {
+ "type": "keyword",
+ "index": true
+ },
+ "lastUpdated": {
+ "type": "date",
+ "format": "date_optional_time"
+ }
+ }
+ }
+}