Fix deletion of redirected URLs

core/src/main/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBolt.java

@@ -70,6 +70,9 @@ public abstract class AbstractStatusUpdaterBolt extends BaseRichBolt {
      */
     public static String roundDateParamName = "status.updater.unit.round.date";
 
+    /** Parameter name to enable deletion of URLs with permanent redirects. */
+    public static String deleteRedirectionsParamName = "status.updater.delete.redirections";
+
     /**
      * Key used to pass a preset Date to use as nextFetchDate. The value must represent a valid
      * instant in UTC and be parsable using {@link DateTimeFormatter#ISO_INSTANT}. This also
@@ -93,6 +96,9 @@ public abstract class AbstractStatusUpdaterBolt extends BaseRichBolt {
 
     private int roundDateUnit = Calendar.SECOND;
 
+    private boolean deleteRedirections = false;
+    private boolean allowRedirs = true;
+
     @Override
     public void prepare(
             Map<String, Object> stormConf, TopologyContext context, OutputCollector collector) {
@@ -103,6 +109,9 @@ public void prepare(
         mdTransfer = MetadataTransfer.getInstance(stormConf);
 
         useCache = ConfUtils.getBoolean(stormConf, useCacheParamName, true);
+        deleteRedirections = ConfUtils.getBoolean(stormConf, deleteRedirectionsParamName, false);
+
+        allowRedirs = ConfUtils.getBoolean(stormConf, Constants.AllowRedirParamName, true);
 
         if (useCache) {
             String spec = ConfUtils.getString(stormConf, cacheConfigParamName);
@@ -118,6 +127,7 @@ public void prepare(
                         return v;
                     },
                     30);
+
             CrawlerMetrics.registerGauge(
                     context,
                     stormConf,
@@ -128,6 +138,7 @@ public void prepare(
                         return v;
                     },
                     30);
+
             CrawlerMetrics.registerGauge(
                     context, stormConf, "cache.size", cache::estimatedSize, 30);
         }
@@ -156,7 +167,7 @@ public void execute(Tuple tuple) {
         // store it again
         if (potentiallyNew && useCache) {
             if (cache.getIfPresent(url) != null) {
-                // no need to add it to the queue
+                // no need to add the URL to the queue
                 LOG.debug("URL {} already in cache", url);
                 cacheHits++;
                 collector.ack(tuple);
@@ -214,15 +225,24 @@ public void execute(Tuple tuple) {
         if (!status.equals(Status.FETCH_ERROR)) {
             metadata.remove(Constants.fetchErrorCountParamName);
         }
+
         // https://github.com/apache/stormcrawler/issues/415
         // remove error related key values in case of success
         if (status.equals(Status.FETCHED) || status.equals(Status.REDIRECTION)) {
             metadata.remove(Constants.STATUS_ERROR_CAUSE);
             metadata.remove(Constants.STATUS_ERROR_MESSAGE);
             metadata.remove(Constants.STATUS_ERROR_SOURCE);
-        } else if (status == Status.ERROR) {
+        }
+
+        if (status == Status.ERROR) {
             // gone? notify any deleters. Doesn't need to be anchored
             collector.emit(Constants.DELETION_STREAM_NAME, new Values(url, metadata));
+        } else if (status == Status.REDIRECTION && deleteRedirections && allowRedirs) {
+            String statusCode = metadata.getFirstValue("fetch.statusCode");
+
+            if ("301".equals(statusCode) || "308".equals(statusCode)) {
+                collector.emit(Constants.DELETION_STREAM_NAME, new Values(url, metadata));
+            }
         }
 
         // determine the value of the next fetch based on the status

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

@@ -307,6 +307,10 @@ config:
   # Can also take "MINUTE" or "HOUR"
   status.updater.unit.round.date: "SECOND"
 
+  # Emit permanently redirected URLs (HTTP 301/308) on the deletion stream
+  # so that they can be removed from the index. Requires redirections.allowed.
+  status.updater.delete.redirections: false
+
   # configuration for the classes extending AbstractIndexerBolt
   # indexer.md.filter: "someKey=aValue"
   indexer.md.docid: ""

core/src/test/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBoltTest.java

@@ -0,0 +1,273 @@
+/*
+ * 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.persistence;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.tuple.Tuple;
+import org.apache.stormcrawler.Constants;
+import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.TestOutputCollector;
+import org.apache.stormcrawler.TestUtil;
+import org.junit.jupiter.api.Test;
+
+class AbstractStatusUpdaterBoltTest {
+
+    @Test
+    void testPermanentRedirect301IsEmittedToDeletionStream() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        Map<String, Object> config = createConfig();
+        config.put(AbstractStatusUpdaterBolt.deleteRedirectionsParamName, true);
+
+        bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/old-page";
+        Metadata metadata = new Metadata();
+        metadata.setValue("fetch.statusCode", "301");
+
+        Tuple tuple = createTuple(url, Status.REDIRECTION, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(1, deletions.size());
+        assertEquals(url, deletions.get(0).get(0));
+
+        Metadata emittedMetadata = (Metadata) deletions.get(0).get(1);
+        assertEquals("301", emittedMetadata.getFirstValue("fetch.statusCode"));
+    }
+
+    @Test
+    void testPermanentRedirect308IsEmittedToDeletionStream() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        Map<String, Object> config = createConfig();
+        config.put(AbstractStatusUpdaterBolt.deleteRedirectionsParamName, true);
+
+        bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/old-page";
+        Metadata metadata = new Metadata();
+        metadata.setValue("fetch.statusCode", "308");
+
+        Tuple tuple = createTuple(url, Status.REDIRECTION, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(1, deletions.size());
+        assertEquals(url, deletions.get(0).get(0));
+
+        Metadata emittedMetadata = (Metadata) deletions.get(0).get(1);
+        assertEquals("308", emittedMetadata.getFirstValue("fetch.statusCode"));
+    }
+
+    @Test
+    void testTemporaryRedirect302IsNotEmittedToDeletionStream() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        Map<String, Object> config = createConfig();
+        config.put(AbstractStatusUpdaterBolt.deleteRedirectionsParamName, true);
+
+        bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/old-page";
+        Metadata metadata = new Metadata();
+        metadata.setValue("fetch.statusCode", "302");
+
+        Tuple tuple = createTuple(url, Status.REDIRECTION, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(0, deletions.size());
+    }
+
+    @Test
+    void testMetaRefreshRedirectIsNotEmittedToDeletionStream() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        Map<String, Object> config = createConfig();
+        config.put(AbstractStatusUpdaterBolt.deleteRedirectionsParamName, true);
+
+        bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/old-page";
+        Metadata metadata = new Metadata();
+        metadata.setValue("fetch.statusCode", "200");
+        metadata.setValue("_redirTo", "http://example.com/new-page");
+
+        Tuple tuple = createTuple(url, Status.REDIRECTION, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(0, deletions.size());
+    }
+
+    @Test
+    void testRedirectionWithoutStatusCodeIsNotEmittedToDeletionStream() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        Map<String, Object> config = createConfig();
+        config.put(AbstractStatusUpdaterBolt.deleteRedirectionsParamName, true);
+
+        bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/old-page";
+        Metadata metadata = new Metadata();
+
+        Tuple tuple = createTuple(url, Status.REDIRECTION, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(0, deletions.size());
+    }
+
+    @Test
+    void testPermanentRedirectIsNotDeletedWhenRedirectionsAreDisabled() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        Map<String, Object> config = createConfig();
+        config.put(AbstractStatusUpdaterBolt.deleteRedirectionsParamName, true);
+        config.put(Constants.AllowRedirParamName, false);
+
+        bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/old-page";
+        Metadata metadata = new Metadata();
+        metadata.setValue("fetch.statusCode", "301");
+
+        Tuple tuple = createTuple(url, Status.REDIRECTION, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(0, deletions.size());
+    }
+
+    @Test
+    void testPermanentRedirectIsNotDeletedByDefault() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        bolt.prepare(
+                createConfig(), TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/old-page";
+        Metadata metadata = new Metadata();
+        metadata.setValue("fetch.statusCode", "301");
+
+        Tuple tuple = createTuple(url, Status.REDIRECTION, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(0, deletions.size());
+    }
+
+    @Test
+    void testFetchedUrlIsNotEmittedToDeletionStream() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        bolt.prepare(
+                createConfig(), TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/page";
+        Metadata metadata = new Metadata();
+        metadata.setValue("fetch.statusCode", "200");
+
+        Tuple tuple = createTuple(url, Status.FETCHED, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(0, deletions.size());
+    }
+
+    @Test
+    void testErrorIsEmittedToDeletionStream() {
+        TestOutputCollector output = new TestOutputCollector();
+        TestStatusUpdaterBolt bolt = new TestStatusUpdaterBolt();
+
+        bolt.prepare(
+                createConfig(), TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        String url = "http://example.com/error";
+        Metadata metadata = new Metadata();
+
+        Tuple tuple = createTuple(url, Status.ERROR, metadata);
+
+        bolt.execute(tuple);
+
+        List<List<Object>> deletions = output.getEmitted(Constants.DELETION_STREAM_NAME);
+
+        assertEquals(1, deletions.size());
+        assertEquals(url, deletions.get(0).get(0));
+    }
+
+    private static Map<String, Object> createConfig() {
+        Map<String, Object> config = new HashMap<>();
+        config.put(AbstractStatusUpdaterBolt.useCacheParamName, false);
+        config.put("scheduler.class", "org.apache.stormcrawler.persistence.DefaultScheduler");
+        return config;
+    }
+
+    private static Tuple createTuple(String url, Status status, Metadata metadata) {
+        Map<String, Object> tupleValues = new HashMap<>();
+        tupleValues.put("url", url);
+        tupleValues.put("status", status);
+        tupleValues.put("metadata", metadata);
+
+        return TestUtil.getMockedTestTuple(tupleValues);
+    }
+
+    private static class TestStatusUpdaterBolt extends AbstractStatusUpdaterBolt {
+
+        @Override
+        protected void store(
+                String url,
+                Status status,
+                Metadata metadata,
+                java.util.Optional<java.util.Date> nextFetch,
+                Tuple tuple) {
+            collector.ack(tuple);
+        }
+    }
+}

docs/src/main/asciidoc/configuration.adoc

@@ -353,6 +353,7 @@ that is being calculated by a link:https://github.com/apache/stormcrawler/blob/m
 | max.fetch.errors | 3 | Maximum number of successive fetch errors before changing status to ERROR.
 | scheduler.class | org.apache.stormcrawler.persistence.DefaultScheduler | Scheduler implementation for computing next fetch dates. Use AdaptiveScheduler for change-rate-based intervals.
 | status.updater.cache.spec | maximumSize=10000, expireAfterAccess=1h | Cache specification for the status updater.
+| status.updater.delete.redirections | false | Whether to emit permanently redirected URLs (HTTP 301/308) on the deletion stream. Requires redirections.allowed.
 | status.updater.unit.round.date | SECOND | Unit for rounding the next fetch date. Can also be MINUTE or HOUR.
 | status.updater.use.cache | true | Whether to use cache to avoid re-persisting URLs.
 |===