GH-3837: Stop SERVICE materialization when query cancellation is observed

[goutamadwant]

Refs #3837

Pull request Description:

Check the outer query's cancellation signal before executing a SERVICE request and while materializing its results. When cancellation is observed, abort the HTTP query and close its response without starting another potentially blocking read.

Successful responses remain fully materialized, preserving SERVICE SILENT error handling and variable remapping. The existing HTTP timeout configuration is unchanged.

Tests cover cancellation before a request, during result consumption, after a blocked read returns, and at the end of results; repeated abort/close; detached results with and without a cancellation signal; SILENT behavior; variable remapping; and timeout propagation.

Validation: all eleven cancellation regression cases, the full ARQ reactor, 97 HTTP integration tests, and 12 Fuseki-main service access tests passed. The integration build skipped UI tooling/tests, Javadocs, and RAT; this is not a full-project verification claim.

This does not interrupt a network read already in progress or address all the Fuseki thread and connection exhaustion factors discussed in #3837. A local execution of the reported workload still stalled after client timeouts. This is a partial cancellation improvement, not a fix for the entire incident, and is not intended to close the issue.

---

• Tests are included.
• Documentation change and updates are provided for the Apache Jena website
• Commits have been squashed to remove intermediate development commit messages.
• Key commit messages start with the issue number (GH-xxxx)

By submitting this pull request, I acknowledge that I am making a contribution to the Apache Software Foundation under the terms and conditions of the Contributor's Agreement.

---

See the Apache Jena "Contributing" guide.

[rvesse]

Thanks for the contribution, this does look to improve the situation for this bug

Couple of comments about potentially simplifying the logic

[rvesse]

I am always a little wary of doing checks like this twice inside a loop although I can understand the reasoning (checking both before and after the network read which could block), might it make more sense to make this a while loop conditional on the cancel signal? Only other change that would necessitate would be a final cancellation check post-loop as hasNext could be false leading to break out of the loop before the next cancellation check

For the case when no cancel signal is available it might also be simpler to preserve existing code path from deleted lines then the actual cancellation check doesn't need to make a null check as well

[goutamadwant]

Thanks @rvesse. restoredc the original materialization path when no cancel signal is available and moved the cancellation guards into a while condition. The final check also catches cancellation when hasNext() returns false. Kept the guards around the potentially blocking read and added coverage for cancellation at the end of results and successful results with and without a signal. let me know if this looks good

[Aklakan]

[Updated proposal] My feeling is that it would be better to defer materialization into its own iterator.

The reason Service.exec consumes data eagerly is that often users would not cleanly close query executions, causing HTTP connection pools to become exhausted which in turn would result in random hangs across an application doing jena queries. So in order to reduce community load, it was decided that Service.exec would always eagerly consume the data and free the HTTP connection.

In fact, some of @goutamadwant tests do not close the QueryIter returned from Service.exec.

My updated proposal is to still do the materialization in its own newnly introduced QueryIter, but use a background thread. The thread ensures that all data will be consumed, whereas the QueryIter wrapping makes it possible to use the usual queryExec.abort() machinery, backed by Thread.interrupt().
Jena transactions are thread-based so an execution involving a separate thread is somewhat dangerous. However, QueryExecHTTP executions don't participate in transactions so in this case it should be fine.

// Service.java
    @Deprecated
    public static QueryIterator exec(OpService op, Context context) {
        ExecutionContext execCxt = ExecutionContext.create(context);
    	return exec(op, context, execCxt);
    }

    public static QueryIterator exec(OpService op, ExecutionContext execCxt) {
        /* ... */

        Context finalContext = context;
        Query finalQuery = query;
        // Build the execution
        Creator<? extends QueryExec> qExecCreator = () -> QueryExecHTTP.newBuilder()
                .endpoint(serviceURL)
                .timeout(timeoutMillis, TimeUnit.MILLISECONDS)
                .httpHeader(HttpNames.hUserAgent, HttpEnv.UserAgent)
                .query(finalQuery)
                .params(serviceParams)
                .context(finalContext)
                .httpClient(httpClient)
                .sendMode(querySendMode)
                .build();

        QueryIterator qIter = new QueryIterThreadedSubExecution(execCxt, qExecCreator);

        // Touch hasNext() to raise exceptions expected by some tests.
        try {
        	qIter.hasNext();
	    } catch (Throwable t) {
	        try {
	            if (t instanceof HttpException ex) {
	                throw QueryExceptionHTTP.rewrap(ex);
	            }
	            throw t;
	        } finally {
	      		qIter.close();
	        }
        }

        if (requiresRemapping)
            qIter = QueryIter.map(qIter, varMapping);

        return qIter;
    }

public class QueryIterThreadedSubExecution
	extends QueryIter
{
	private final Creator<? extends QueryExec> queryExecCreator;
	private final BlockingDeque<Elt> elts = new LinkedBlockingDeque<>();
	private final Thread thread;
	private Binding peekedBinding = null;

	// Only start emitting bindings after the producer thread has terminated.
	// This may prevent deadlocks due to resource exhaustion (e.g. HTTP connection pools).
	private boolean awaitThreadTermination = true;

	public QueryIterThreadedSubExecution(ExecutionContext execCxt, Creator<? extends QueryExec> queryExecCreator) {
		super(execCxt);
		this.queryExecCreator = queryExecCreator;
		this.thread = new Thread(this::run);
		thread.start();
	}

	private void ensurePeekedBinding() {
		Elt peekedElt;
		if (peekedBinding == null) {
			try {
				peekedElt = elts.take();
			} catch (InterruptedException | QueryCancelledException e) {
				throw new QueryCancelledException(e);
			}

			Throwable t = peekedElt.throwable;
			if (t != null) {
				if (t instanceof InterruptedException || t instanceof QueryCancelledException) {
					throw new QueryCancelledException(t);
				} else if (t instanceof HttpException e) {
		            throw QueryExceptionHTTP.rewrap(e);
				} else if (t instanceof QueryExceptionHTTP e) {
					throw e;
				} else {
					throw new QueryException(t);
				}
			}
			peekedBinding = peekedElt.binding;
		}
	}

	@Override
	protected boolean hasNextBinding() {
		ensurePeekedBinding();
		return peekedBinding != POISON;
	}

	@Override
	protected Binding moveToNextBinding() {
		ensurePeekedBinding();

		if (awaitThreadTermination) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				throw new QueryCancelledException(e);
			}
		}

		Binding result;
		if (peekedBinding == POISON) {
			throw new NoSuchElementException();
		}

		result = peekedBinding;
		peekedBinding = null;
		return result;
	}

	@Override
	protected void requestCancel() {
		thread.interrupt();
		elts.notifyAll();
	}

	@Override
	protected void closeIterator() {
		thread.interrupt();
		Duration timeout = Duration.ofSeconds(10);
		try {
			thread.join(timeout);
		} catch (InterruptedException e) {
			throw new ARQException("Abandoned thread which did not terminate within " + timeout);
		}
	}

	// Producer thread logic.

	private record Elt(Binding binding, Throwable throwable) {}
	private Binding POISON = BindingFactory.binding(Var.alloc("__POISON___"), NodeFactory.createBlankNode());

	private void run() {
		ExecutionContext execCxt = getExecContext();
		try (QueryExec queryExec = queryExecCreator.create()) {
			RowSet rs = queryExec.select();
			while (rs.hasNext()) {
				execCxt.checkCancelSignal();
				Binding binding = rs.next();
				elts.put(new Elt(binding, null));
			}
			forcePut(elts, new Elt(POISON, null));
		} catch (Exception e) {
			forcePut(elts, new Elt(null, e));
		}
	}

	private static <T> void forcePut(BlockingQueue<T> queue, T item) {
		retry: while (true) {
			try {
				queue.put(item);
				break;
			} catch (InterruptedException e) {
				try {
					Thread.sleep(100);
				} catch (InterruptedException e2) {
					// Ignore
				}
				continue retry;
			}
		}
	}
}

This seems to work with the existing tests. Some of @goutamadwant tests assume that Service.exec would eagerly materialize and raise exceptions before returning. Modifying those tests to expect failure during iterator consumption makes those work.

This code could be used in this PR (feel free to use it). If preferred, I could also make a separate PR proposal from it.

[Aklakan]

I now refined/update my previous comment and tested against slightly modified tests of this PR according to description.