AssertionError thrown from AbstractSailConnection when using Federation

[ieurich]

When adding multiple repositories to a federation and evaluating a query which contains a NaryJoin the query is executed over multiple threads. This can result in AbstractSailConnection.startUpdate throwing an assertion error because multiple threads attempt to flush the same federation connection instance concurrently. This also results in the query failing to return the expected results.

Here is an example test which reproduces the problem. On my machine this test reproduces the issue maybe once in ten runs on average.

public void testAssertionErrorReproduction() {

        final ValueFactory vf = SimpleValueFactory.getInstance();
        final BNode address = vf.createBNode();
        final BNode anotherAddress = vf.createBNode();

        final ModelBuilder builder = new ModelBuilder();
        builder
            .setNamespace("ex", "http://example.org/")
            .subject("ex:Picasso")
                .add(RDF.TYPE, "ex:Artist")
                .add(FOAF.FIRST_NAME, "Pablo")
                .add("ex:homeAddress", address)
            .subject("ex:AnotherArtist") 
                .add(RDF.TYPE, "ex:Artist")
                .add(FOAF.FIRST_NAME, "AnotherArtist")
                .add("ex:homeAddress", anotherAddress)
            .subject(address)           
                .add("ex:street", "31 Art Gallery")
                .add("ex:city", "Madrid")
                .add("ex:country", "Spain")
            .subject(anotherAddress)           
                .add("ex:street", "32 Art Gallery")
                .add("ex:city", "London")
                .add("ex:country", "UK");

        final Model model = builder.build();
        final Repository repo1 = new SailRepository(new MemoryStore());
        repo1.initialize();
        repo1.getConnection().add(model);

        final Repository repo2 = new SailRepository(new MemoryStore());
        repo2.initialize();

        final Federation fed = new Federation();
        fed.addMember(repo1);
        fed.addMember(repo2);

        final String ex = "http://example.org/";
        final String queryString =
                "PREFIX rdf: <" + RDF.NAMESPACE + ">\n" +
                "PREFIX foaf: <" + FOAF.NAMESPACE + ">\n" +
                "PREFIX ex: <" + ex + ">\n" +  
                "select (count(?persons) as ?count) {\n" +
                "   ?persons rdf:type ex:Artist ;\n"
                + "          ex:homeAddress ?country .\n"
                + " ?country ex:country \"Spain\" . }";

        final SailRepository fedRepo = new SailRepository(fed);
        fedRepo.initialize();

        final SailRepositoryConnection fedRepoConn = fedRepo.getConnection();

        fedRepoConn.begin();
        final TupleQuery query = fedRepoConn.prepareTupleQuery(QueryLanguage.SPARQL, queryString);
        final TupleQueryResult eval = query.evaluate();
        if (eval.hasNext()) {
            final Value next = eval.next().getValue("count");
            assertEquals(1, ((Literal) next).intValue());
        }
        else {
            fail("No result");
        }

        fedRepoConn.commit();
    }

Here is the stack trace.

Exception in thread "rdf4j-federation-1" java.lang.AssertionError at org.eclipse.rdf4j.sail.helpers.AbstractSailConnection.startUpdate(AbstractSailConnection.java:512) at org.eclipse.rdf4j.sail.helpers.AbstractSailConnection.flush(AbstractSailConnection.java:388) at org.eclipse.rdf4j.sail.helpers.AbstractSailConnection.flushPendingUpdates(AbstractSailConnection.java:929) at org.eclipse.rdf4j.sail.helpers.AbstractSailConnection.getStatements(AbstractSailConnection.java:318) at org.eclipse.rdf4j.sail.federation.AbstractFederationConnection$FederationTripleSource.getStatements(AbstractFederationConnection.java:400) at org.eclipse.rdf4j.query.algebra.evaluation.impl.StrictEvaluationStrategy.evaluate(StrictEvaluationStrategy.java:538) at org.eclipse.rdf4j.query.algebra.evaluation.impl.StrictEvaluationStrategy.evaluate(StrictEvaluationStrategy.java:234) at org.eclipse.rdf4j.sail.federation.evaluation.FederationStrategy.evaluate(FederationStrategy.java:65) at org.eclipse.rdf4j.sail.federation.evaluation.ParallelJoinCursor.run(ParallelJoinCursor.java:82) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)

[hmottestad]

I have managed to reproduce your issue by putting your code inside a loop. I then modified your code to correctly close all closeable objects and the issue is solved. See: https://github.com/eclipse/rdf4j/pull/1619

[hmottestad]

        @Test
    public void testAssertionErrorReproduction() {

        for (int i = 0; i < 100; i++) {

            ValueFactory vf = SimpleValueFactory.getInstance();
            BNode address = vf.createBNode();
            BNode anotherAddress = vf.createBNode();

            ModelBuilder builder = new ModelBuilder();
            builder
                    .setNamespace("ex", "http://example.org/")
                    .subject("ex:Picasso")
                    .add(RDF.TYPE, "ex:Artist")
                    .add(FOAF.FIRST_NAME, "Pablo")
                    .add("ex:homeAddress", address)
                    .subject("ex:AnotherArtist")
                    .add(RDF.TYPE, "ex:Artist")
                    .add(FOAF.FIRST_NAME, "AnotherArtist")
                    .add("ex:homeAddress", anotherAddress)
                    .subject(address)
                    .add("ex:street", "31 Art Gallery")
                    .add("ex:city", "Madrid")
                    .add("ex:country", "Spain")
                    .subject(anotherAddress)
                    .add("ex:street", "32 Art Gallery")
                    .add("ex:city", "London")
                    .add("ex:country", "UK");

            Model model = builder.build();
            Repository repo1 = new SailRepository(new MemoryStore());
            repo1.init();
            try (RepositoryConnection connection = repo1.getConnection()) {
                connection.add(model);
            }

            Repository repo2 = new SailRepository(new MemoryStore());
            repo2.init();

            Federation fed = new Federation();
            fed.addMember(repo1);
            fed.addMember(repo2);

            String ex = "http://example.org/";
            String queryString = "PREFIX rdf: <" + RDF.NAMESPACE + ">\n" +
                    "PREFIX foaf: <" + FOAF.NAMESPACE + ">\n" +
                    "PREFIX ex: <" + ex + ">\n" +
                    "select (count(?persons) as ?count) {\n" +
                    "   ?persons rdf:type ex:Artist ;\n"
                    + "          ex:homeAddress ?country .\n"
                    + " ?country ex:country \"Spain\" . }";

            SailRepository fedRepo = new SailRepository(fed);
            fedRepo.init();

            try (SailRepositoryConnection fedRepoConn = fedRepo.getConnection()) {

                fedRepoConn.begin();
                TupleQuery query = fedRepoConn.prepareTupleQuery(QueryLanguage.SPARQL, queryString);
                try (TupleQueryResult eval = query.evaluate()) {
                    if (eval.hasNext()) {
                        Value next = eval.next().getValue("count");
                        assertEquals(1, ((Literal) next).intValue());
                    } else {
                        fail("No result");
                    }
                }

                fedRepoConn.commit();
            }
        }
    }

[hmottestad]

fyi repos should be shutdown() too

[ieurich]

Thanks for taking a look. However, I tried a few runs of your code and it failed with the same stack trace again. Could you take another look please?

[hmottestad]

I upped the loop to 1000000 and it is still passing. Which version of RDF4J are you using?

[hmottestad]

Reproduced on develop branch.

[hmottestad]

Affected line.

[hmottestad]

@ieurich thank you for your persistence :)

[hmottestad]

@ieurich this has now been fixed and will be released as part of 3.0.2.

I work on RDF4J on my spare time for free, so if you or your organisation finds RDF4J useful then I would appreciate if you considered sponsoring me on Github: https://github.com/sponsors/hmottestad

[hmottestad]

@ieurich 3.0.2 is released with a fix for this issue: https://rdf4j.eclipse.org/news/2019/10/26/rdf4j-3.0.2-released/