Allow indexer bolts to use a metadata value as the document ID

core/src/main/java/org/apache/stormcrawler/indexing/AbstractIndexerBolt.java

@@ -75,8 +75,32 @@ public abstract class AbstractIndexerBolt extends BaseRichBolt {
     /** Indicates that empty field values should not be emitted at all. */
     public static final String ignoreEmptyFieldValueParamName = "indexer.ignore.empty.fields";
 
+    /**
+     * Metadata key whose value should be used to derive the document ID, if present. The value is
+     * passed through the same SHA-256 digest used for the URL-based ID, so this can be used e.g. by
+     * a {@link org.apache.stormcrawler.parse.ParseFilter} generating a hash of the content so that
+     * duplicate content ends up under the same document ID regardless of the URL(s) it was found
+     * at. Falls back to the default (a SHA-256 digest of the URL) if not set or if the metadata
+     * does not contain a value for the configured key.
+     *
+     * <p><strong>Deletion path caveat:</strong> a {@code DeletionBolt} handling a "gone" page does
+     * not reparse it, so it only sees this key if it was included in {@code metadata.persist} (see
+     * {@code MetadataTransfer#metadataPersistParamName}) and therefore round-tripped through the
+     * status store. If the key is missing from {@code metadata.persist}, the deletion falls back to
+     * {@code sha256(url)}, which will not match the ID the document was indexed under, silently
+     * orphaning it.
+     *
+     * <p><strong>Content-dedup caveat:</strong> when several URLs resolve to the same
+     * metadata-based ID (e.g. a content hash shared across duplicate pages), deleting any one of
+     * those URLs deletes the shared document, even though the content may still be live under the
+     * other URLs, until one of them is re-fetched and re-indexes it.
+     */
+    public static final String DOC_ID_METADATA_PARAM_NAME = "indexer.md.docid";
+
     private String[] filterKeyValue = null;
 
+    private String docIdMetadataKey = null;
+
     private final List<Key> metadata2field = new ArrayList<>();
 
     private String fieldNameForText = null;
@@ -157,6 +181,8 @@ public void prepare(
 
         canonicalMetadataName = ConfUtils.getString(conf, canonicalMetadataParamName);
 
+        docIdMetadataKey = ConfUtils.getString(conf, DOC_ID_METADATA_PARAM_NAME);
+
         final Pattern indexValuePattern = Pattern.compile("\\[(\\d+)\\]");
 
         for (String mapping : ConfUtils.loadListFromConf(metadata2fieldParamName, conf)) {
@@ -259,10 +285,28 @@ protected Map<String, String[]> filterMetadata(Metadata meta) {
      *
      * @param metadata The {@link Metadata}.
      * @param normalisedUrl The normalised url.
-     * @return Return the normalised url SHA-256 digest as String.
+     * @return The SHA-256 digest of the value found in the metadata for the key configured with
+     *     {@link #DOC_ID_METADATA_PARAM_NAME}, if any, otherwise the SHA-256 digest of the
+     *     normalised url.
      */
     protected String getDocumentID(Metadata metadata, String normalisedUrl) {
-        return org.apache.commons.codec.digest.DigestUtils.sha256Hex(normalisedUrl);
+        final String fromMetadata = getDocumentIDFromMetadata(metadata);
+        final String source = fromMetadata != null ? fromMetadata : normalisedUrl;
+        return org.apache.commons.codec.digest.DigestUtils.sha256Hex(source);
+    }
+
+    /**
+     * Returns the value found in the metadata for the key configured with {@link
+     * #DOC_ID_METADATA_PARAM_NAME}, or null if the parameter is not set or has no matching value in
+     * the metadata. Subclasses which override {@link #getDocumentID} entirely (e.g. to hash or
+     * sanitize the ID for a specific backend) should call this first so that they still honor the
+     * configured metadata-based override.
+     */
+    protected final String getDocumentIDFromMetadata(Metadata metadata) {
+        if (docIdMetadataKey == null) {
+            return null;
+        }
+        return StringUtils.trimToNull(metadata.getFirstValue(docIdMetadataKey));
     }
 
     /**

core/src/main/resources/crawler-default.yaml

@@ -303,6 +303,7 @@ config:
 
   # configuration for the classes extending AbstractIndexerBolt
   # indexer.md.filter: "someKey=aValue"
+  indexer.md.docid: ""
   indexer.ignore.empty.fields: false
   indexer.url.fieldname: "url"
   indexer.text.fieldname: "content"

core/src/test/java/org/apache/stormcrawler/indexer/BasicIndexingTest.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.TreeSet;
+import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.stormcrawler.Metadata;
 import org.apache.stormcrawler.indexing.AbstractIndexerBolt;
 import org.junit.jupiter.api.Assertions;
@@ -197,6 +198,44 @@ void testEmptyFilterMetadata() throws Exception {
                 "Index only the URL if no mapping is provided");
     }
 
+    @Test
+    void testDocumentIDFromMetadata() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+        config.put(AbstractIndexerBolt.DOC_ID_METADATA_PARAM_NAME, "content.hash");
+        Metadata metadata = new Metadata();
+        metadata.setValue("content.hash", "abcdef123456");
+        prepareIndexerBolt(config);
+        String docId = ((DummyIndexer) bolt).docId(metadata, URL);
+        Assertions.assertEquals(
+                DigestUtils.sha256Hex("abcdef123456"),
+                docId,
+                "The SHA-256 digest of the value found in the metadata should be used as the"
+                        + " document ID");
+    }
+
+    @Test
+    void testDocumentIDFromMetadataMissingValue() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+        config.put(AbstractIndexerBolt.DOC_ID_METADATA_PARAM_NAME, "content.hash");
+        prepareIndexerBolt(config);
+        String docId = ((DummyIndexer) bolt).docId(new Metadata(), URL);
+        Assertions.assertEquals(
+                DigestUtils.sha256Hex(URL),
+                docId,
+                "Should fall back to the URL digest if the metadata key has no value");
+    }
+
+    @Test
+    void testDocumentIDDefault() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+        prepareIndexerBolt(config);
+        String docId = ((DummyIndexer) bolt).docId(new Metadata(), URL);
+        Assertions.assertEquals(
+                DigestUtils.sha256Hex(URL),
+                docId,
+                "Should default to the URL digest if no metadata key is configured");
+    }
+
     @Test
     void testGlobFilterMetadata() throws Exception {
         Map<String, Object> config = new HashMap<>();

core/src/test/java/org/apache/stormcrawler/indexer/DummyIndexer.java

@@ -86,4 +86,8 @@ private String trimValue(String value) {
     public Map<String, String> returnFields() {
         return fields;
     }
+
+    public String docId(Metadata metadata, String url) {
+        return getDocumentID(metadata, url);
+    }
 }

docs/src/main/asciidoc/configuration.adoc

@@ -318,13 +318,23 @@ The values below are used by sub-classes of `AbstractIndexerBolt`.
 
 | indexer.canonical.name | canonical | Metadata key for the canonical URL. Used to replace the URL with its canonical form before indexing.
 | indexer.ignore.empty.fields | false | If true, skip fields with empty values when indexing.
+| indexer.md.docid | - | Metadata key whose value, if present, is SHA-256 digested and used as the document ID instead of the default SHA-256 digest of the URL. Useful e.g. for content-based deduplication when a `ParseFilter` stores a content hash in the metadata.
 | indexer.md.filter | - | YAML list of key=value filters for metadata-based indexing.
 | indexer.md.mapping | - | YAML mapping from metadata fields to persistence layer fields.
 | indexer.text.fieldname | content | Field name for indexed HTML body text.
 | indexer.text.maxlength | -1 | Maximum length of text to index. -1 means no limit.
 | indexer.url.fieldname | url | Field name for indexed URL.
 |===
 
+Note: `indexer.md.docid` is never used raw: the matching value is passed through SHA-256
+(the same digest used for the default URL-based ID), except for CloudSearch, whose
+`CloudSearchIndexerBolt` re-digests it with SHA-512 via `CloudSearchUtils.getID` to satisfy
+CloudSearch's ID constraints. It is only honoured by backends whose bolts call
+`AbstractIndexerBolt#getDocumentID` under the hood: the OpenSearch and OpenSearch-Java
+`IndexerBolt`/`DeletionBolt` pairs, and the CloudSearch `CloudSearchIndexerBolt`. The Solr and
+SQL indexers build their document/row key directly from the URL and never call
+`getDocumentID()`, so this setting is silently ignored for those two backends.
+
 ==== Status Persistence
 
 This refers to persisting the status of a URL (e.g. ERROR, DISCOVERED etc.) along with something like a `nextFetchDate`

external/aws/src/main/java/org/apache/stormcrawler/aws/bolt/CloudSearchIndexerBolt.java

@@ -174,6 +174,10 @@ public void prepare(
 
     @Override
     protected String getDocumentID(Metadata metadata, String normalisedUrl) {
+        final String fromMetadata = getDocumentIDFromMetadata(metadata);
+        if (fromMetadata != null) {
+            return CloudSearchUtils.getID(fromMetadata);
+        }
         return CloudSearchUtils.getID(normalisedUrl);
     }
 

external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/bolt/DeletionBolt.java

@@ -19,12 +19,14 @@
 
 import java.lang.invoke.MethodHandles;
 import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
 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.Metadata;
+import org.apache.stormcrawler.indexing.AbstractIndexerBolt;
 import org.apache.stormcrawler.metrics.CrawlerMetrics;
 import org.apache.stormcrawler.opensearch.AsyncBulkProcessor;
 import org.apache.stormcrawler.opensearch.OpenSearchConnection;
@@ -56,6 +58,8 @@ public class DeletionBolt extends BaseRichBolt implements AsyncBulkProcessor.Lis
 
     private WaitAckCache waitAck;
 
+    private String docIdMetadataKey;
+
     public DeletionBolt() {}
 
     /** Sets the index name instead of taking it from the configuration. * */
@@ -71,6 +75,9 @@ public void prepare(
             indexName = ConfUtils.getString(conf, IndexerBolt.OSIndexNameParamName, "content");
         }
 
+        docIdMetadataKey =
+                ConfUtils.getString(conf, AbstractIndexerBolt.DOC_ID_METADATA_PARAM_NAME);
+
         try {
             connection = OpenSearchConnection.getConnection(conf, BOLT_TYPE, this);
         } catch (Exception e1) {
@@ -121,14 +128,25 @@ protected String getIndexName(Metadata m) {
     }
 
     /**
-     * Get the document id.
+     * Get the document id. Uses the SHA-256 digest of the value found in the metadata for the key
+     * configured with {@link AbstractIndexerBolt#DOC_ID_METADATA_PARAM_NAME} if present, so that
+     * the ID matches the one generated by the indexer for the same document.
      *
      * @param metadata The {@link Metadata}.
      * @param url The normalised url.
-     * @return Return the normalised url SHA-256 digest as String.
+     * @return Return the SHA-256 digest of the value found in the metadata, or of the normalised
+     *     url if not present.
      */
     protected String getDocumentID(Metadata metadata, String url) {
-        return org.apache.commons.codec.digest.DigestUtils.sha256Hex(url);
+        String source = url;
+        if (docIdMetadataKey != null) {
+            final String fromMetadata =
+                    StringUtils.trimToNull(metadata.getFirstValue(docIdMetadataKey));
+            if (fromMetadata != null) {
+                source = fromMetadata;
+            }
+        }
+        return org.apache.commons.codec.digest.DigestUtils.sha256Hex(source);
     }
 
     @Override

external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/bolt/DeletionBolt.java

@@ -19,12 +19,14 @@
 
 import java.lang.invoke.MethodHandles;
 import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
 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.Metadata;
+import org.apache.stormcrawler.indexing.AbstractIndexerBolt;
 import org.apache.stormcrawler.metrics.CrawlerMetrics;
 import org.apache.stormcrawler.opensearch.OpenSearchConnection;
 import org.apache.stormcrawler.opensearch.WaitAckCache;
@@ -56,6 +58,8 @@ public class DeletionBolt extends BaseRichBolt implements Listener {
 
     private WaitAckCache waitAck;
 
+    private String docIdMetadataKey;
+
     public DeletionBolt() {}
 
     /** Sets the index name instead of taking it from the configuration. * */
@@ -71,6 +75,9 @@ public void prepare(
             indexName = ConfUtils.getString(conf, IndexerBolt.OSIndexNameParamName, "content");
         }
 
+        docIdMetadataKey =
+                ConfUtils.getString(conf, AbstractIndexerBolt.DOC_ID_METADATA_PARAM_NAME);
+
         try {
             connection = OpenSearchConnection.getConnection(conf, BOLT_TYPE, this);
         } catch (Exception e1) {
@@ -119,14 +126,25 @@ protected String getIndexName(Metadata m) {
     }
 
     /**
-     * Get the document id.
+     * Get the document id. Uses the SHA-256 digest of the value found in the metadata for the key
+     * configured with {@link AbstractIndexerBolt#DOC_ID_METADATA_PARAM_NAME} if present, so that
+     * the ID matches the one generated by the indexer for the same document.
      *
      * @param metadata The {@link Metadata}.
      * @param url The normalised url.
-     * @return Return the normalised url SHA-256 digest as String.
+     * @return Return the SHA-256 digest of the value found in the metadata, or of the normalised
+     *     url if not present.
      */
     protected String getDocumentID(Metadata metadata, String url) {
-        return org.apache.commons.codec.digest.DigestUtils.sha256Hex(url);
+        String source = url;
+        if (docIdMetadataKey != null) {
+            final String fromMetadata =
+                    StringUtils.trimToNull(metadata.getFirstValue(docIdMetadataKey));
+            if (fromMetadata != null) {
+                source = fromMetadata;
+            }
+        }
+        return org.apache.commons.codec.digest.DigestUtils.sha256Hex(source);
     }
 
     @Override