You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stormcrawler.apache.org by jn...@apache.org on 2025/10/23 07:10:13 UTC
(stormcrawler) branch main updated: #1698 OpenSearch: Make waitAck cache configurable via cache specifica… (#1700)
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
The following commit(s) were added to refs/heads/main by this push:
new 38fa1a09 #1698 OpenSearch: Make waitAck cache configurable via cache specifica… (#1700)
38fa1a09 is described below
commit 38fa1a09685fb7ec5a45cb5b24d98a8521814cdc
Author: Harry <11...@users.noreply.github.com>
AuthorDate: Thu Oct 23 16:10:08 2025 +0900
#1698 OpenSearch: Make waitAck cache configurable via cache specifica… (#1700)
* #1698 OpenSearch: Make waitAck cache configurable via cache specification
This commit replaces the hardcoded waitAck Caffeine cache configuration
in StatusUpdaterBolt with a configurable cache specification string.
It introduces the parameter `opensearch.status.waitack.cache.spec`,
allowing users to control cache behavior (size, expiration) via config.
* #1698 OpenSearch: add TestCase
* #1698 OpenSearch: add TestCase
* #1698 OpenSearch: code formatting
* #1698 apply code cleanup and config adjustments
- update expireAfterWrite duration
- avoid star imports
- restore previous test code
- remove Korean comments
- delete unclear test case
* #1698 align waitAck cache TTL with Storm message timeout
* #1698 remove unnecessary maximumSize limit from waitAck cache
* #1698 specify Locale.ROOT in String.format to avoid forbidden API violation
* #1698 Revert import statement changes
---
external/opensearch/opensearch-conf.yaml | 4 ++++
.../opensearch/persistence/StatusUpdaterBolt.java | 16 ++++++++++------
2 files changed, 14 insertions(+), 6 deletions(-)
diff --git a/external/opensearch/opensearch-conf.yaml b/external/opensearch/opensearch-conf.yaml
index c78c3ce1..6836df11 100644
--- a/external/opensearch/opensearch-conf.yaml
+++ b/external/opensearch/opensearch-conf.yaml
@@ -114,6 +114,10 @@ config:
opensearch.status.recentDate.increase: -1
opensearch.status.recentDate.min.gap: -1
+ # Caffeine cache specification for the waitAck cache used in StatusUpdaterBolt.
+ # If not set, the value of topology.message.timeout.secs is used for expireAfterWrite (default: 300s)
+ # opensearch.status.waitack.cache.spec: "maximumSize=10000,expireAfterWrite=300s"
+
topology.metrics.consumer.register:
- class: "org.apache.stormcrawler.opensearch.metrics.MetricsConsumer"
parallelism.hint: 1
diff --git a/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java b/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java
index d4ce11ca..c16083a7 100644
--- a/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java
+++ b/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java
@@ -31,7 +31,6 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import org.apache.commons.lang.StringUtils;
@@ -160,11 +159,16 @@ public class StatusUpdaterBolt extends AbstractStatusUpdaterBolt
fieldNameForRoutingKey = fieldNameForRoutingKey.replaceAll("\\.", "%2E");
}
- waitAck =
- Caffeine.newBuilder()
- .expireAfterWrite(60, TimeUnit.SECONDS)
- .removalListener(this)
- .build();
+ String defaultSpec =
+ String.format(
+ Locale.ROOT,
+ "expireAfterWrite=%ds",
+ ConfUtils.getInt(stormConf, "topology.message.timeout.secs", 300));
+
+ String waitAckSpec =
+ ConfUtils.getString(stormConf, "opensearch.status.waitack.cache.spec", defaultSpec);
+
+ waitAck = Caffeine.from(waitAckSpec).removalListener(this).build();
int metrics_time_bucket_secs = 30;