#2282 Don't apply SitemapFilter to redirect targets

core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java

@@ -262,7 +262,7 @@ public void run() {
                         continue;
                     }
 
-                    final Metadata fetchMetadata = metadata;
+                    final Metadata fetchMetadata = redirectMetadata(metadata);
                     ProtocolResponse response;
                     response =
                             fetchHelpers.call(
@@ -316,7 +316,7 @@ && allowRedirs()
                                     fit.tuple(),
                                     url,
                                     outcome.redirectTarget(),
-                                    outcome.metadata(),
+                                    redirectMetadata(outcome.metadata()),
                                     outcome.redirectKeyValues());
                         }
                         collector.emit(

core/src/main/java/org/apache/stormcrawler/bolt/JSoupParserBolt.java

@@ -437,7 +437,8 @@ public void execute(Tuple tuple) {
 
                     // https://github.com/apache/stormcrawler/issues/954
                     if (allowRedirs() && StringUtils.isNotBlank(redirection)) {
-                        emitOutlink(tuple, URLUtil.toURL(url), redirection, metadata);
+                        emitOutlink(
+                                tuple, URLUtil.toURL(url), redirection, redirectMetadata(metadata));
                     }
 
                     // Mark URL as redirected

core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java

@@ -449,7 +449,7 @@ public void execute(Tuple input) {
 
             long start = System.currentTimeMillis();
             final String fetchUrl = urlString;
-            final Metadata fetchMetadata = metadata;
+            final Metadata fetchMetadata = redirectMetadata(metadata);
             ProtocolResponse response;
             response =
                     fetchHelpers.call(
@@ -555,11 +555,11 @@ public void execute(Tuple input) {
                                 input,
                                 url,
                                 redirection,
-                                mergedMetadata,
+                                redirectMetadata(mergedMetadata),
                                 SiteMapParserBolt.isSitemapKey,
                                 "true");
                     } else {
-                        emitOutlink(input, url, redirection, mergedMetadata);
+                        emitOutlink(input, url, redirection, redirectMetadata(mergedMetadata));
                     }
                 }
                 // Mark URL as redirected

core/src/main/java/org/apache/stormcrawler/bolt/StatusEmitterBolt.java

@@ -119,6 +119,21 @@ protected boolean allowRedirs() {
         return allowRedirs;
     }
 
+    /**
+     * Returns the metadata to filter the redirect target of a URL with: a copy without {@link
+     * SiteMapParserBolt#foundSitemapKey} if the metadata has it. SitemapFilter drops the outlinks
+     * of a page on a site with sitemaps, which must not apply to where the URL itself redirects to.
+     */
+    protected static Metadata redirectMetadata(Metadata metadata) {
+        if (metadata.getFirstValue(SiteMapParserBolt.foundSitemapKey) == null) {
+            return metadata;
+        }
+        Metadata copy = new Metadata();
+        copy.putAll(metadata);
+        copy.remove(SiteMapParserBolt.foundSitemapKey);
+        return copy;
+    }
+
     @Override
     public void cleanup() {
         if (urlFilters != null) {

core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java

@@ -19,9 +19,11 @@
 
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlMatching;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
 import static org.awaitility.Awaitility.await;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -52,6 +54,8 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 @WireMockTest
 abstract class AbstractFetcherBoltTest {
@@ -444,6 +448,78 @@ void nonNumericMetricsHeaderDoesNotFailTheFetch(WireMockRuntimeInfo wmRuntimeInf
         Assertions.assertEquals(0, output.getEmitted(Constants.StatusStreamName).size());
     }
 
+    /**
+     * With SitemapFilter in the chain, the redirect target of the fetched URL is discovered whether
+     * robots.txt declares a sitemap or not: the filter is meant for outlinks of non-sitemap pages.
+     */
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void redirectTargetIsDiscoveredWithSitemapFilter(
+            boolean robotsDeclaresSitemap, WireMockRuntimeInfo wmRuntimeInfo)
+            throws ReflectiveOperationException {
+        String robots = "User-agent: *\n";
+        if (robotsDeclaresSitemap) {
+            robots += "Sitemap: " + wmRuntimeInfo.getHttpBaseUrl() + "/sitemap.xml\n";
+        }
+        stubFor(get(urlEqualTo("/robots.txt")).willReturn(aResponse().withBody(robots)));
+        String target = wmRuntimeInfo.getHttpBaseUrl() + "/target";
+        stubFor(
+                get(urlEqualTo("/start"))
+                        .willReturn(aResponse().withStatus(301).withHeader("Location", target)));
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("http.agent.name", "this_is_only_a_test");
+        config.put("http.allow.redirects", false);
+        config.put("redirections.allowed", true);
+        config.put("sitemap.discovery", true);
+        config.put("urlfilters.config.file", "urlfilters-sitemap-only.json");
+
+        TestOutputCollector output = fetch(wmRuntimeInfo, config, "/start");
+        List<List<Object>> statusTuples = output.getEmitted(Constants.StatusStreamName);
+        Assertions.assertTrue(
+                statusTuples.stream()
+                        .anyMatch(t -> t.get(0).equals(target) && t.get(2) == Status.DISCOVERED),
+                "redirect target not discovered, got "
+                        + statusTuples.stream().map(t -> t.get(0) + " " + t.get(2)).toList());
+    }
+
+    /**
+     * Same as above with http.allow.redirects=true: the protocol follows the redirect itself and
+     * filters each hop, so SitemapFilter must not reject the target either.
+     */
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void redirectIsFollowedByProtocolWithSitemapFilter(
+            boolean robotsDeclaresSitemap, WireMockRuntimeInfo wmRuntimeInfo)
+            throws ReflectiveOperationException {
+        String robots = "User-agent: *\n";
+        if (robotsDeclaresSitemap) {
+            robots += "Sitemap: " + wmRuntimeInfo.getHttpBaseUrl() + "/sitemap.xml\n";
+        }
+        stubFor(get(urlEqualTo("/robots.txt")).willReturn(aResponse().withBody(robots)));
+        String target = wmRuntimeInfo.getHttpBaseUrl() + "/target";
+        stubFor(
+                get(urlEqualTo("/start"))
+                        .willReturn(aResponse().withStatus(301).withHeader("Location", target)));
+        stubFor(get(urlEqualTo("/target")).willReturn(aResponse().withBody("target page")));
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("http.agent.name", "this_is_only_a_test");
+        config.put("http.allow.redirects", true);
+        config.put("sitemap.discovery", true);
+        config.put("urlfilters.config.file", "urlfilters-sitemap-only.json");
+
+        TestOutputCollector output = fetch(wmRuntimeInfo, config, "/start");
+        verify(1, getRequestedFor(urlEqualTo("/target")));
+        List<List<Object>> parseTuples = output.getEmitted(Utils.DEFAULT_STREAM_ID);
+        Assertions.assertEquals(1, parseTuples.size());
+        // the fetched page itself keeps foundSitemap for SitemapFilter on its outlinks
+        Metadata metadata = (Metadata) parseTuples.get(0).get(2);
+        Assertions.assertEquals(
+                Boolean.toString(robotsDeclaresSitemap),
+                metadata.getFirstValue(SiteMapParserBolt.foundSitemapKey));
+    }
+
     TestOutputCollector fetch(
             WireMockRuntimeInfo wmRuntimeInfo, Map<String, Object> config, String path)
             throws ReflectiveOperationException {

core/src/test/java/org/apache/stormcrawler/bolt/JSoupParserBoltTest.java

@@ -236,6 +236,28 @@ void testHTMLRedir() throws IOException {
         Assertions.assertEquals(2, statusTuples.size());
     }
 
+    /**
+     * The target of a meta refresh is discovered with SitemapFilter in the chain even if the site
+     * declares a sitemap: the filter is meant for outlinks of non-sitemap pages.
+     */
+    @Test
+    void testHTMLRedirWithSitemapFilter() throws IOException {
+        stormConf.put("sitemap.discovery", true);
+        stormConf.put("urlfilters.config.file", "urlfilters-sitemap-only.json");
+        bolt.prepare(stormConf, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+        Metadata metadata = new Metadata();
+        metadata.setValue(SiteMapParserBolt.foundSitemapKey, "true");
+        parse("http://www.somesite.com", "redir.html", metadata);
+        List<List<Object>> statusTuples = output.getEmitted(Constants.StatusStreamName);
+        Assertions.assertEquals(2, statusTuples.size());
+        Assertions.assertTrue(
+                statusTuples.stream()
+                        .anyMatch(
+                                t ->
+                                        t.get(0).equals("http://www.somesite.com/site")
+                                                && t.get(2) == Status.DISCOVERED));
+    }
+
     @Test
     void testExecuteWithOutlinksLimit() throws IOException {
         stormConf.put("parser.emitOutlinks.max.per.page", 5);

core/src/test/java/org/apache/stormcrawler/filtering/sitemap/SitemapFilterTest.java

@@ -0,0 +1,91 @@
+/*
+ * 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.filtering.sitemap;
+
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.bolt.SiteMapParserBolt;
+import org.apache.stormcrawler.filtering.URLFilter;
+import org.apache.stormcrawler.util.URLUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class SitemapFilterTest {
+
+    private static final String OUTLINK = "http://www.example.com/page";
+
+    private URLFilter createFilter(boolean discovery) {
+        SitemapFilter filter = new SitemapFilter();
+        Map<String, Object> conf = new HashMap<>();
+        conf.put("sitemap.discovery", discovery);
+        filter.configure(conf, JsonNodeFactory.instance.objectNode(), "SitemapFilter");
+        return filter;
+    }
+
+    private static Metadata foundSitemap() {
+        Metadata metadata = new Metadata();
+        metadata.setValue(SiteMapParserBolt.foundSitemapKey, "true");
+        return metadata;
+    }
+
+    private static URL source() throws MalformedURLException {
+        return URLUtil.toURL("http://www.example.com/");
+    }
+
+    @Test
+    void keptWhenDiscoveryIsOff() throws MalformedURLException {
+        URLFilter filter = createFilter(false);
+        Assertions.assertEquals(OUTLINK, filter.filter(source(), foundSitemap(), OUTLINK));
+    }
+
+    @Test
+    void keptWhenTheSiteHasNoSitemap() throws MalformedURLException {
+        URLFilter filter = createFilter(true);
+        Metadata metadata = new Metadata();
+        metadata.setValue(SiteMapParserBolt.foundSitemapKey, "false");
+        Assertions.assertEquals(OUTLINK, filter.filter(source(), metadata, OUTLINK));
+    }
+
+    /** An outlink of a page which is not a sitemap, on a site which declares sitemaps. */
+    @Test
+    void outlinkOfNonSitemapPageIsDropped() throws MalformedURLException {
+        URLFilter filter = createFilter(true);
+        Assertions.assertNull(filter.filter(source(), foundSitemap(), OUTLINK));
+    }
+
+    @Test
+    void outlinkOfSitemapIsKept() throws MalformedURLException {
+        URLFilter filter = createFilter(true);
+        Metadata metadata = foundSitemap();
+        metadata.setValue(SiteMapParserBolt.isSitemapKey, "true");
+        Assertions.assertEquals(OUTLINK, filter.filter(source(), metadata, OUTLINK));
+    }
+
+    /** sitemap.discovery in the metadata overrides the configuration. */
+    @Test
+    void discoveryTurnedOffInMetadataKeepsTheOutlink() throws MalformedURLException {
+        URLFilter filter = createFilter(true);
+        Metadata metadata = foundSitemap();
+        metadata.setValue("sitemap.discovery", "false");
+        Assertions.assertEquals(OUTLINK, filter.filter(source(), metadata, OUTLINK));
+    }
+}

core/src/test/resources/urlfilters-sitemap-only.json

@@ -0,0 +1,8 @@
+{
+	"org.apache.stormcrawler.filtering.URLFilters": [
+		{
+			"class": "org.apache.stormcrawler.filtering.sitemap.SitemapFilter",
+			"name": "SitemapFilter"
+		}
+	]
+}