HostURLFilter memoizes the source host in plain fields shared by all fetcher threads

[rzo1]

What happens

HostURLFilter caches the last source URL and its host and domain in three plain instance fields and compares by identity to decide whether the cache applies. StatusEmitterBolt.prepare() builds one URLFilters instance per bolt, and FetcherBolt runs fetcher.threads.number fetcher threads (10 by default) through it, so the three fields are written and read concurrently with no synchronisation. The write of previousSourceUrl is not atomic with the writes of the host and domain, so one thread can see its own URL in previousSourceUrl next to another thread's host in previousSourceHost and compare the candidate against the wrong origin.

Where

core/src/main/java/org/apache/stormcrawler/filtering/host/HostURLFilter.java:48-50 and 94-107.

} else {
    fromHost = sourceUrl.getHost();
    ...
    previousSourceHost = fromHost;
    previousSourceDomain = fromDomain;
    previousSourceUrl = sourceUrl;
}

Why it matters

Both directions of the wrong answer are possible: a URL outside the source host is admitted, or a same-host URL is dropped. The reachable path is narrow. FetcherBolt builds a fresh URL object per fetch and emits at most one redirect outlink at FetcherBolt.java:857, so the identity check misses there and the redirect is always evaluated against its own source. The loop over robots-discovered sitemaps at FetcherBolt.java:633-643 calls emitOutlink repeatedly with the same URL object, which is where the memoization is actually used, and it is not steerable by whoever supplies the sitemap list. Even so, the state is shared across threads without any memory barrier, and the memoization saves only one PaidLevelDomain.getPLD call.

Reproduction

Save as core/src/test/java/org/apache/stormcrawler/filtering/HostURLFilterConcurrencyTest.java.

/*
 * 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;

import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.net.URL;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.stormcrawler.Metadata;
import org.apache.stormcrawler.filtering.host.HostURLFilter;
import org.apache.stormcrawler.util.URLUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

/**
 * One HostURLFilter instance is shared by all FetcherBolt threads through URLFilters, so its
 * memoization of the source host is read and written concurrently.
 */
class HostURLFilterConcurrencyTest {

    private static final int ITERATIONS = 2_000_000;

    @Test
    void sharedInstanceNeverAdmitsACrossHostUrl() throws Exception {
        HostURLFilter filter = new HostURLFilter();
        ObjectNode params = new ObjectNode(JsonNodeFactory.instance);
        params.put("ignoreOutsideHost", Boolean.TRUE);
        params.put("ignoreOutsideDomain", Boolean.FALSE);
        Map<String, Object> conf = new HashMap<>();
        filter.configure(conf, params);

        final URL sourceA = URLUtil.toURL("http://a.example.com/page");
        final URL sourceB = URLUtil.toURL("http://b.example.org/page");
        final Metadata metadata = new Metadata();

        final AtomicLong admittedCrossHost = new AtomicLong();
        final AtomicLong droppedSameHost = new AtomicLong();
        final CountDownLatch start = new CountDownLatch(1);

        // asks about a host that is never its own
        Thread a =
                new Thread(
                        () -> {
                            await(start);
                            for (int i = 0; i < ITERATIONS; i++) {
                                if (filter.filter(sourceA, metadata, "http://b.example.org/x")
                                        != null) {
                                    admittedCrossHost.incrementAndGet();
                                }
                            }
                        });

        // asks about its own host
        Thread b =
                new Thread(
                        () -> {
                            await(start);
                            for (int i = 0; i < ITERATIONS; i++) {
                                if (filter.filter(sourceB, metadata, "http://b.example.org/y")
                                        == null) {
                                    droppedSameHost.incrementAndGet();
                                }
                            }
                        });

        a.start();
        b.start();
        start.countDown();
        a.join();
        b.join();

        Assertions.assertEquals(
                0L,
                admittedCrossHost.get(),
                "cross-host URLs admitted out of " + ITERATIONS + " evaluations");
        Assertions.assertEquals(
                0L,
                droppedSameHost.get(),
                "same-host URLs dropped out of " + ITERATIONS + " evaluations");
    }

    private static void await(CountDownLatch latch) {
        try {
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

Run it:

mvn -pl core test -Dtest=HostURLFilterConcurrencyTest

Two threads share one filter with ignoreOutsideHost: true; one asks about a host that is never its own, the other asks about its own host. This is a stress test rather than a deterministic one: it failed on every run here, but the number of bad answers varies from run to run and from machine to machine, so treat the count below as an order of magnitude and not as a fixture.

[ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 0.882 s
[ERROR]   HostURLFilterConcurrencyTest.sharedInstanceNeverAdmitsACrossHostUrl:90
    cross-host URLs admitted out of 2000000 evaluations ==> expected: <0> but was: <7579>

Other runs on the same machine reported counts between roughly 6000 and 11000. Every run failed.

Suggested fix

Drop the previousSourceUrl / previousSourceHost / previousSourceDomain fields from HostURLFilter.filter() and recompute the host and the paid-level domain from sourceUrl on each call. If the PaidLevelDomain.getPLD call turns out to be worth caching, hold the three values in one immutable holder object behind a single volatile field, so a reader sees a consistent triple. Keep the stress test at a smaller iteration count if two million evaluations is too slow for the normal test run.