From 61322aca4d45c1a7c994e9b4f225c4a7d801e475 Mon Sep 17 00:00:00 2001 From: imbajin Date: Mon, 12 Oct 2026 14:25:42 +0800 Subject: [PATCH 1/2] fix(core): close stopped traversal iterators - close the prepared iterator on every consumer exit - keep the original failure and suppress closing errors - verify real native cursor ownership after caller cleanup --- .../traversal/algorithm/OltpTraverser.java | 36 ++- .../apache/hugegraph/core/CoreTestSuite.java | 1 + .../core/OltpTraverserCursorTest.java | 237 ++++++++++++++++++ 3 files changed, 261 insertions(+), 13 deletions(-) create mode 100644 hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java index 9d12b0ee9a..fbc4da281c 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java @@ -333,20 +333,30 @@ public EdgesConsumer(Consumer consumer, long capacity) { @Override public void accept(Iterator edgeIter) { - Iterator ids = prepare(edgeIter); - long counter = 0; - while (ids.hasNext()) { - if (Thread.currentThread().isInterrupted()) { - LOG.warn("Consumer is Interrupted"); - break; + Iterator toClose = edgeIter; + Throwable failure = null; + try { + Iterator ids = prepare(edgeIter); + toClose = ids != null ? ids : edgeIter; + long counter = 0; + while (ids.hasNext()) { + if (Thread.currentThread().isInterrupted()) { + LOG.warn("Consumer is Interrupted"); + break; + } + counter++; + this.consumer.accept(ids.next()); } - counter++; - this.consumer.accept(ids.next()); - } - long total = edgeIterCounter.addAndGet(counter); - // traverse by batch & improve performance - if (this.capacity != NO_LIMIT && total >= this.capacity) { - throw new Consumers.StopExecution("reach capacity"); + long total = edgeIterCounter.addAndGet(counter); + // traverse by batch & improve performance + if (this.capacity != NO_LIMIT && total >= this.capacity) { + throw new Consumers.StopExecution("reach capacity"); + } + } catch (RuntimeException | Error e) { + failure = e; + throw e; + } finally { + closeIterator(toClose, failure); } } } diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java index 3d6d3149fc..d3b308bd6b 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java @@ -38,6 +38,7 @@ @RunWith(Suite.class) @Suite.SuiteClasses({ BackendLeaseCleanupTest.class, + OltpTraverserCursorTest.class, PropertyKeyCoreTest.class, VertexLabelCoreTest.class, EdgeLabelCoreTest.class, diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java new file mode 100644 index 0000000000..05f0f0ca8d --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java @@ -0,0 +1,237 @@ +/* + * 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.hugegraph.core; + +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Iterator; +import java.util.Set; +import java.util.UUID; + +import org.apache.commons.configuration2.MapConfiguration; +import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.HugeFactory; +import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.backend.query.QueryBatch; +import org.apache.hugegraph.backend.query.QueryResults; +import org.apache.hugegraph.backend.store.rocksdb.RocksDBSessions; +import org.apache.hugegraph.backend.tx.GraphTransaction; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.id.Id; +import org.apache.hugegraph.id.IdGenerator; +import org.apache.hugegraph.iterator.MapperIterator; +import org.apache.hugegraph.query.Query; +import org.apache.hugegraph.testutil.Utils; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.traversal.algorithm.OltpTraverser; +import org.apache.hugegraph.type.define.Directions; +import org.apache.hugegraph.util.Consumers; +import org.apache.tinkerpop.gremlin.structure.Edge; +import org.apache.tinkerpop.gremlin.structure.T; +import org.apache.tinkerpop.gremlin.structure.Vertex; +import org.apache.tinkerpop.gremlin.structure.util.CloseableIterator; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Before; +import org.junit.Test; +import org.rocksdb.RocksIterator; + +public class OltpTraverserCursorTest { + + @Before + public void requireRocksDB() { + Assume.assumeTrue("Native cursor regression requires RocksDB", + "rocksdb".equals(Utils.getConf().getString("backend"))); + CoreTestSuite.initEnv(); + } + + @Test + public void testConsumerFailureClosesNativeCursor() throws Exception { + this.checkNativeCursor("consumer"); + } + + @Test + public void testStopExecutionClosesNativeCursor() throws Exception { + this.checkNativeCursor("stop"); + } + + @Test + public void testInterruptedConsumerClosesNativeCursor() throws Exception { + this.checkNativeCursor("interrupt"); + } + + @Test + public void testPrepareFailureClosesNativeCursor() throws Exception { + this.checkNativeCursor("prepare"); + } + + @Test + public void testCloseFailureIsSuppressedAfterNativeCursorCloses() throws Exception { + this.checkNativeCursor("close"); + } + + private void checkNativeCursor(String outcome) throws Exception { + Path directory = Files.createTempDirectory("oltp-cursor-"); + HugeConfig config = new HugeConfig(new MapConfiguration(Collections.emptyMap())); + config.setProperty("backend", "rocksdb"); + config.setProperty("serializer", "binary"); + config.setProperty("store", "cursor_" + UUID.randomUUID().toString().replace("-", "")); + config.setProperty("oltp.concurrent_threads", 0); + config.setProperty("rocksdb.data_path", directory.toString()); + config.setProperty("rocksdb.wal_path", directory.toString()); + HugeGraph graph = null; + RocksDBSessions control = null; + Iterator edges = null; + RocksIterator nativeCursor = null; + try { + graph = HugeFactory.open(config); + graph.initBackend(); + graph.schema().propertyKey("seq").asInt().create(); + graph.schema().vertexLabel("vertex").useCustomizeNumberId().create(); + graph.schema().edgeLabel("linked").sourceLabel("vertex").targetLabel("vertex") + .multiTimes().properties("seq").sortKeys("seq").create(); + Vertex source = graph.addVertex(T.label, "vertex", T.id, 1L); + Vertex target = graph.addVertex(T.label, "vertex", T.id, 2L); + // The backend prefetches 500 edges; record 501 keeps its native cursor open. + for (int i = 0; i < 501; i++) { + source.addEdge("linked", target, "seq", i); + } + graph.tx().commit(); + Id label = graph.schema().getEdgeLabel("linked").id(); + Query query = GraphTransaction.constructEdgesQuery(IdGenerator.of(1L), Directions.OUT, + new Id[]{label}); + edges = graph.edges(query); + Assert.assertTrue(edges.hasNext()); + nativeCursor = nativeCursor(edges, Collections.newSetFromMap(new IdentityHashMap<>())); + Assert.assertNotNull("Expected the real RocksDB scan cursor", nativeCursor); + Assert.assertTrue(nativeCursor.isOwningHandle()); + Object store = graph.storeProvider().loadGraphStore(config); + RocksDBSessions sessions = Whitebox.getInternalState(store, "sessions"); + // Keep the native DB alive during handle observation after caller FINISHED cleanup. + control = sessions.copy(config, "cursor-control", "g"); + control.session(); + Probe traverser = new Probe(graph); + RuntimeException failure = "stop".equals(outcome) ? new Consumers.StopExecution("limit") : + new IllegalStateException("business failure"); + if ("close".equals(outcome)) { + edges = new MapperIterator(edges, edge -> edge) { + @Override + public void close() throws Exception { + super.close(); + throw new IllegalStateException("close failure"); + } + }; + } + Iterator input = edges; + OltpTraverser.EdgesConsumer consumer; + if ("prepare".equals(outcome)) { + consumer = traverser.new EdgesConsumer(id -> Assert.fail(), OltpTraverser.NO_LIMIT) { + @Override + protected Iterator prepare(Iterator iterator) { + throw failure; + } + }; + } else { + consumer = traverser.new OneStepEdgeIterConsumer(id -> { + throw failure; + }, OltpTraverser.NO_LIMIT); + } + if ("interrupt".equals(outcome)) { + Thread.currentThread().interrupt(); + try { + consumer.accept(input); + Assert.assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + } else { + RuntimeException thrown = org.apache.hugegraph.testutil.Assert.assertThrows( + RuntimeException.class, () -> consumer.accept(input)); + Assert.assertSame(failure, thrown); + if ("close".equals(outcome)) { + Assert.assertEquals(1, thrown.getSuppressed().length); + Assert.assertTrue(thrown.getSuppressed()[0].getCause().getMessage().contains("close failure")); + } + } + HugeFactory.closeCurrentThreadTransactions(); + Assert.assertTrue("Control lease must keep the DB alive", control.databaseOpened()); + Assert.assertFalse("Traversal retained its native cursor after " + outcome, + nativeCursor.isOwningHandle()); + } finally { + // These are teardown only; the native handle assertion above must run first. + try { + if (nativeCursor != null && nativeCursor.isOwningHandle()) { + nativeCursor.close(); + } + if (edges != null && !"close".equals(outcome)) { + CloseableIterator.closeIterator(edges); + } + } finally { + try { + HugeFactory.closeCurrentThreadTransactions(); + if (control != null) { + control.close(); + } + if (graph != null && !graph.closed()) { + graph.drop(); + } + } finally { + if (graph != null) { + HugeFactory.remove(graph); + } + FileUtils.deleteDirectory(directory.toFile()); + } + } + } + } + + private static RocksIterator nativeCursor(Object value, Set visited) throws Exception { + if (value instanceof RocksIterator) { + return (RocksIterator) value; + } + if (!(value instanceof Iterator || value instanceof QueryResults || value instanceof QueryBatch) || + !visited.add(value)) { + return null; + } + for (Class type = value.getClass(); type != null && + type.getName().startsWith("org.apache.hugegraph."); type = type.getSuperclass()) { + for (Field field : type.getDeclaredFields()) { + if (Modifier.isStatic(field.getModifiers())) { + continue; + } + field.setAccessible(true); + RocksIterator found = nativeCursor(field.get(value), visited); + if (found != null) { + return found; + } + } + } + return null; + } + + private static class Probe extends OltpTraverser { + + Probe(HugeGraph graph) { + super(graph); + } + } +} From 0198608461d7102519ad7cb1bf90576d16d2a496 Mon Sep 17 00:00:00 2001 From: imbajin Date: Mon, 12 Oct 2026 18:59:59 +0800 Subject: [PATCH 2/2] fix(core): stop producers after traversal stop - propagate clean consumer stop to producer loops - retain terminal business and cleanup failures - verify bounded lazy production with real native cursors --- .../org/apache/hugegraph/util/Consumers.java | 1 + .../core/OltpTraverserCursorTest.java | 79 ++++++++++++++++++- .../hugegraph/unit/util/ConsumersTest.java | 47 ++++++++++- 3 files changed, 121 insertions(+), 6 deletions(-) diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java index 1a8a12b443..914d4161fe 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java @@ -268,6 +268,7 @@ public void provide(V v) throws Throwable { if (this.exception != null) { throw this.throwException(); } + throw new StopExecution("Consumer stopped"); } } catch (InterruptedException e) { LOG.warn("Interrupt while queuing QUEUE_END", e); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java index 51ff2adcbf..3e53074f8e 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/OltpTraverserCursorTest.java @@ -130,19 +130,25 @@ public void testAsyncFatalErrorReleasesBlockedProducer() throws Exception { this.checkNativeCursor("async-fatal-producer"); } + @Test + public void testCapacityStopEndsActualLazyTraversal() throws Exception { + this.checkNativeCursor("lazy-producer"); + } + private void checkNativeCursor(String outcome) throws Exception { Path directory = Files.createTempDirectory("oltp-cursor-"); HugeConfig config = new HugeConfig(new MapConfiguration(Collections.emptyMap())); config.setProperty("backend", "rocksdb"); config.setProperty("serializer", "binary"); config.setProperty("store", "cursor_" + UUID.randomUUID().toString().replace("-", "")); - config.setProperty("oltp.concurrent_threads", 0); + config.setProperty("oltp.concurrent_threads", "lazy-producer".equals(outcome) ? 1 : 0); config.setProperty("rocksdb.data_path", directory.toString()); config.setProperty("rocksdb.wal_path", directory.toString()); HugeGraph graph = null; RocksDBSessions control = null; Iterator edges = null; RocksIterator nativeCursor = null; + int producerReads = -1; try { graph = HugeFactory.open(config); graph.initBackend(); @@ -170,6 +176,10 @@ private void checkNativeCursor(String outcome) throws Exception { // Keep the native DB alive during handle observation after caller FINISHED cleanup. control = sessions.copy(config, "cursor-control", "g"); control.session(); + if ("lazy-producer".equals(outcome)) { + // This isolated JVM fixture owns the single-worker pool used below. + OltpTraverser.destroy(); + } Probe traverser = new Probe(graph); Throwable failure = outcome.contains("fatal") ? new AssertionError("fatal failure") : "stop".equals(outcome) ? new Consumers.StopExecution("limit") : @@ -198,15 +208,19 @@ protected Iterator prepare(Iterator iterator) { }; } else { consumer = traverser.new OneStepEdgeIterConsumer(id -> { - if (!outcome.startsWith("async-capacity")) { + if (!outcome.startsWith("async-capacity") && !"lazy-producer".equals(outcome)) { if (failure instanceof Error) { throw (Error) failure; } throw (RuntimeException) failure; } - }, outcome.startsWith("async-capacity") ? 1L : OltpTraverser.NO_LIMIT); + }, outcome.startsWith("async-capacity") || "lazy-producer".equals(outcome) ? + 1L : OltpTraverser.NO_LIMIT); } - if (outcome.startsWith("async-")) { + if ("lazy-producer".equals(outcome)) { + producerReads = this.consumeLazyProducer(traverser, consumer, input); + Assert.assertEquals(501L, traverser.edgeIterCounter.get()); + } else if (outcome.startsWith("async-")) { AtomicReference handled = new AtomicReference<>(); Throwable thrown = this.consumeAsync(consumer, input, handled, outcome); if (outcome.startsWith("async-capacity")) { @@ -250,6 +264,9 @@ protected Iterator prepare(Iterator iterator) { Assert.assertTrue("Control lease must keep the DB alive", control.databaseOpened()); Assert.assertFalse("Traversal retained its native cursor after " + outcome, nativeCursor.isOwningHandle()); + if ("lazy-producer".equals(outcome)) { + Assert.assertEquals("Capacity Stop must end the lazy producer", 2, producerReads); + } } finally { // These are teardown only; the native handle assertion above must run first. try { @@ -319,6 +336,8 @@ private Throwable consumeAsync(OltpTraverser.EdgesConsumer consumer, if (outcome.contains("producer")) { consumers.provide(Collections.emptyIterator()); } + } catch (Consumers.StopExecution ignored) { + // Stop ends production successfully, as in Oltp's producer loop. } catch (Throwable providedFailure) { failure = providedFailure; } @@ -353,6 +372,54 @@ private Throwable consumeAsync(OltpTraverser.EdgesConsumer consumer, } } + private int consumeLazyProducer(Probe traverser, OltpTraverser.EdgesConsumer consumer, + Iterator input) throws Exception { + Consumers.ExecutorPool pool = Whitebox.getInternalState(OltpTraverser.class, "executors"); + ExecutorService worker = pool.getExecutor(); + pool.returnExecutor(worker); + class LazyInputs implements Iterator>, AutoCloseable { + + private int reads; + private boolean closed; + + @Override + public boolean hasNext() { + if (this.reads == 1) { + try { + // Runs after the real Consumers task and its worker transaction cleanup. + worker.submit(() -> { }).get(5L, TimeUnit.SECONDS); + } catch (Exception e) { + throw new AssertionError(e); + } + } + return this.reads < 100; + } + + @Override + public Iterator next() { + if (this.reads >= 100) { + throw new java.util.NoSuchElementException(); + } + return this.reads++ == 0 ? input : Collections.emptyIterator(); + } + + @Override + public void close() { + this.closed = true; + } + } + LazyInputs source = new LazyInputs(); + try { + long total = traverser.traverseLazy(source, consumer); + Assert.assertEquals((long) source.reads, total); + Assert.assertTrue("Traversal must close its lazy producer", source.closed); + return source.reads; + } finally { + OltpTraverser.destroy(); + Assert.assertTrue(worker.awaitTermination(5L, TimeUnit.SECONDS)); + } + } + private static RocksIterator nativeCursor(Object value, Set visited) throws Exception { if (value instanceof RocksIterator) { return (RocksIterator) value; @@ -382,5 +449,9 @@ private static class Probe extends OltpTraverser { Probe(HugeGraph graph) { super(graph); } + + long traverseLazy(Iterator> source, java.util.function.Consumer> consumer) { + return this.traverseByOne(source, consumer, "native-lazy-producer-stop"); + } } } diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java index f1b4de410c..9db801be19 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java @@ -207,8 +207,12 @@ private void checkTerminalWithFullQueue(boolean blockedProducer, boolean cleanup Assert.assertTrue(calling.await(5L, TimeUnit.SECONDS)); Assert.assertThrows(TimeoutException.class, () -> result.get(100L, TimeUnit.MILLISECONDS)); release.countDown(); - Assert.assertSame(cleanupFailure ? closeFailure : terminalFailure, - result.get(3L, TimeUnit.SECONDS)); + Throwable returned = result.get(3L, TimeUnit.SECONDS); + if (blockedProducer && !cleanupFailure && terminalFailure == null) { + Assert.assertTrue(returned instanceof Consumers.StopExecution); + } else { + Assert.assertSame(cleanupFailure ? closeFailure : terminalFailure, returned); + } Assert.assertTrue(finished.await(5L, TimeUnit.SECONDS)); if (blockedProducer && (cleanupFailure || terminalFailure != null)) { // A failure reported by provide() must not leave await() blocked. @@ -320,6 +324,45 @@ private void checkProducerFailureAfterCleanup(int workers) throws Exception { } } + @Test + public void testCleanStopEndsLazyProducer() throws Exception { + ExecutorService worker = Executors.newFixedThreadPool(1); + CountDownLatch stopped = new CountDownLatch(1); + AtomicInteger reads = new AtomicInteger(); + AtomicInteger processed = new AtomicInteger(); + try { + Consumers consumers = new Consumers<>(worker, value -> { + processed.incrementAndGet(); + throw new Consumers.StopExecution("capacity"); + }, stopped::countDown, 1); + consumers.start("lazy-producer-stop"); + try { + for (int i = 0; i < 100; i++) { + if (i == 1) { + Assert.assertTrue(stopped.await(5L, TimeUnit.SECONDS)); + } + reads.incrementAndGet(); + consumers.provide(i); + } + } catch (Consumers.StopExecution ignored) { + // Same successful producer termination contract as Oltp traversal. + } catch (Throwable failure) { + throw new AssertionError(failure); + } finally { + try { + consumers.await(); + } catch (Throwable failure) { + throw new AssertionError(failure); + } + } + Assert.assertEquals(2, reads.get()); + Assert.assertEquals(1, processed.get()); + } finally { + worker.shutdownNow(); + Assert.assertTrue(worker.awaitTermination(5L, TimeUnit.SECONDS)); + } + } + private static void awaitGate(CountDownLatch gate) { try { if (!gate.await(5L, TimeUnit.SECONDS)) {