This is an automated email from the ASF dual-hosted git repository.
imbajin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hugegraph.git
The following commit(s) were added to refs/heads/master by this push:
new a79a71d64 fix(server): close shortest path edge iterators (#3167)
a79a71d64 is described below
commit a79a71d644eaca2156403841913b4c6ada5f6597
Author: contrueCT <[email protected]>
AuthorDate: Tue Aug 25 01:59:04 2026 +0800
fix(server): close shortest path edge iterators (#3167)
ShortestPathTraverser could return after finding a path without closing the
current backend edge iterator. With HStore, the abandoned iterator could leave
its remote scanner active and Store threads waiting to deliver pages that the
client no longer consumed.
The same ownership gap also existed in the backward search and in the
iterator used by the super-node probe.
---------
Co-authored-by: imbajin <[email protected]>
---
.../hugegraph/iterator/ExtendableIterator.java | 17 +-
.../apache/hugegraph/iterator/LimitIterator.java | 13 +-
.../unit/iterator/ExtendableIteratorTest.java | 33 ++
.../hugegraph/unit/iterator/LimitIteratorTest.java | 20 +
.../traversal/algorithm/HugeTraverser.java | 50 ++-
.../traversal/algorithm/ShortestPathTraverser.java | 111 +++---
.../org/apache/hugegraph/unit/UnitTestSuite.java | 2 +
.../unit/traversal/ShortestPathTraverserTest.java | 429 +++++++++++++++++++++
8 files changed, 614 insertions(+), 61 deletions(-)
diff --git
a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/iterator/ExtendableIterator.java
b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/iterator/ExtendableIterator.java
index b692ad361..5aac08f5e 100644
---
a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/iterator/ExtendableIterator.java
+++
b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/iterator/ExtendableIterator.java
@@ -67,11 +67,26 @@ public class ExtendableIterator<T> extends
WrappedIterator<T> {
@Override
public void close() throws Exception {
+ Throwable failure = null;
for (Iterator<T> iter : this.itors) {
if (iter instanceof AutoCloseable) {
- ((AutoCloseable) iter).close();
+ try {
+ ((AutoCloseable) iter).close();
+ } catch (Exception | Error e) {
+ if (failure == null) {
+ failure = e;
+ } else if (failure != e) {
+ failure.addSuppressed(e);
+ }
+ }
}
}
+ if (failure instanceof Exception) {
+ throw (Exception) failure;
+ }
+ if (failure != null) {
+ throw (Error) failure;
+ }
}
@Override
diff --git
a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/iterator/LimitIterator.java
b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/iterator/LimitIterator.java
index 9df6299a6..9b3739d91 100644
---
a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/iterator/LimitIterator.java
+++
b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/iterator/LimitIterator.java
@@ -24,6 +24,7 @@ public class LimitIterator<T> extends WrappedIterator<T> {
private final Iterator<T> originIterator;
private final Function<T, Boolean> filterCallback;
+ private boolean originClosed;
public LimitIterator(Iterator<T> origin, Function<T, Boolean> filter) {
this.originIterator = origin;
@@ -35,6 +36,15 @@ public class LimitIterator<T> extends WrappedIterator<T> {
return this.originIterator;
}
+ @Override
+ public void close() throws Exception {
+ if (this.originClosed || this.originIterator == null) {
+ return;
+ }
+ this.originClosed = true;
+ super.close();
+ }
+
@Override
protected final boolean fetch() {
while (this.originIterator.hasNext()) {
@@ -56,9 +66,10 @@ public class LimitIterator<T> extends WrappedIterator<T> {
}
protected final void closeOriginIterator() {
- if (this.originIterator == null) {
+ if (this.originClosed || this.originIterator == null) {
return;
}
+ this.originClosed = true;
close(this.originIterator);
}
}
diff --git
a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/iterator/ExtendableIteratorTest.java
b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/iterator/ExtendableIteratorTest.java
index 5e08f2d9c..fe3a55456 100644
---
a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/iterator/ExtendableIteratorTest.java
+++
b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/iterator/ExtendableIteratorTest.java
@@ -162,6 +162,30 @@ public class ExtendableIteratorTest extends BaseUnitTest {
Assert.assertTrue(c3.closed());
}
+ @Test
+ public void testCloseAllWhenCloseFails() {
+ Exception firstFailure = new Exception("first");
+ Exception lastFailure = new Exception("last");
+ CloseableItor<Integer> c1 = new CloseableItor<>(DATA1.iterator(),
+ firstFailure);
+ CloseableItor<Integer> c2 = new CloseableItor<>(DATA2.iterator());
+ CloseableItor<Integer> c3 = new CloseableItor<>(DATA3.iterator(),
+ lastFailure);
+ ExtendableIterator<Integer> results = new ExtendableIterator<>();
+ results.extend(c1).extend(c2).extend(c3);
+
+ Throwable actual = Assert.assertThrows(
+ Exception.class,
+ (Assert.ThrowableRunnable) results::close);
+
+ Assert.assertSame(firstFailure, actual);
+ Assert.assertTrue(c1.closed());
+ Assert.assertTrue(c2.closed());
+ Assert.assertTrue(c3.closed());
+ Assert.assertEquals(1, actual.getSuppressed().length);
+ Assert.assertSame(lastFailure, actual.getSuppressed()[0]);
+ }
+
@Test
public void testCloseAfterNext1() throws Exception {
CloseableItor<Integer> c1 = new CloseableItor<>(DATA1.iterator());
@@ -217,10 +241,16 @@ public class ExtendableIteratorTest extends BaseUnitTest {
AutoCloseable {
private final Iterator<V> iter;
+ private final Exception closeFailure;
private boolean closed = false;
public CloseableItor(Iterator<V> iter) {
+ this(iter, null);
+ }
+
+ public CloseableItor(Iterator<V> iter, Exception closeFailure) {
this.iter = iter;
+ this.closeFailure = closeFailure;
}
@Override
@@ -236,6 +266,9 @@ public class ExtendableIteratorTest extends BaseUnitTest {
@Override
public void close() throws Exception {
this.closed = true;
+ if (this.closeFailure != null) {
+ throw this.closeFailure;
+ }
}
public boolean closed() {
diff --git
a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/iterator/LimitIteratorTest.java
b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/iterator/LimitIteratorTest.java
index f087e7df3..8ee2fcd38 100644
---
a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/iterator/LimitIteratorTest.java
+++
b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/iterator/LimitIteratorTest.java
@@ -178,4 +178,24 @@ public class LimitIteratorTest extends BaseUnitTest {
results.close();
Assert.assertTrue(vals.closed());
}
+
+ @Test
+ public void testCloseOnlyOnceWhenReachLimit() throws Exception {
+ AtomicInteger closeCount = new AtomicInteger();
+ CloseableItor<Integer> vals =
+ new CloseableItor<Integer>(DATA.iterator()) {
+ @Override
+ public void close() throws Exception {
+ closeCount.incrementAndGet();
+ super.close();
+ }
+ };
+ LimitIterator<Integer> results = new LimitIterator<>(vals,
+ val -> true);
+
+ Assert.assertFalse(results.hasNext());
+ results.close();
+
+ Assert.assertEquals(1, closeCount.get());
+ }
}
diff --git
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java
index 8122c7908..fe8fa0568 100644
---
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java
+++
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java
@@ -393,18 +393,7 @@ public class HugeTraverser {
if (labels == null || labels.isEmpty()) {
return this.edgesOfVertex(source, dir, (Id) null, limit);
}
- ExtendableIterator<Edge> results = new ExtendableIterator<>();
- for (Id label : labels.keySet()) {
- E.checkNotNull(label, "edge label");
- results.extend(this.edgesOfVertex(source, dir, label, limit));
- }
-
- if (limit == NO_LIMIT) {
- return results;
- }
-
- long[] count = new long[1];
- return new LimitIterator<>(results, e -> count[0]++ >= limit);
+ return this.edgesOfVertexByLabels(source, dir, labels.keySet(), limit);
}
protected Iterator<Edge> edgesOfVertex(Id source, Directions dir,
@@ -412,10 +401,27 @@ public class HugeTraverser {
if (labels == null || labels.isEmpty()) {
return this.edgesOfVertex(source, dir, (Id) null, limit);
}
+ return this.edgesOfVertexByLabels(source, dir, labels, limit);
+ }
+
+ private Iterator<Edge> edgesOfVertexByLabels(Id source, Directions dir,
+ Iterable<Id> labels,
+ long limit) {
ExtendableIterator<Edge> results = new ExtendableIterator<>();
- for (Id label : labels) {
- E.checkNotNull(label, "edge label");
- results.extend(this.edgesOfVertex(source, dir, label, limit));
+ List<Iterator<Edge>> opened = new ArrayList<>();
+ try {
+ for (Id label : labels) {
+ E.checkNotNull(label, "edge label");
+ Iterator<Edge> edges = this.edgesOfVertex(source, dir,
+ label, limit);
+ opened.add(edges);
+ results.extend(edges);
+ }
+ } catch (RuntimeException | Error e) {
+ for (Iterator<Edge> edges : opened) {
+ closeIterator(edges, e);
+ }
+ throw e;
}
if (limit == NO_LIMIT) {
@@ -426,6 +432,20 @@ public class HugeTraverser {
return new LimitIterator<>(results, e -> count[0]++ >= limit);
}
+ protected static void closeIterator(Iterator<?> iterator,
+ Throwable failure) {
+ try {
+ CloseableIterator.closeIterator(iterator);
+ } catch (RuntimeException | Error closeFailure) {
+ if (failure == null) {
+ throw closeFailure;
+ }
+ if (failure != closeFailure) {
+ failure.addSuppressed(closeFailure);
+ }
+ }
+ }
+
protected Iterator<Edge> edgesOfVertex(Id source, EdgeStep edgeStep) {
if (edgeStep.properties() == null || edgeStep.properties().isEmpty()) {
Iterator<Edge> edges = this.edgesOfVertex(source,
diff --git
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/ShortestPathTraverser.java
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/ShortestPathTraverser.java
index 4a2f3259d..f69fd7da5 100644
---
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/ShortestPathTraverser.java
+++
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/ShortestPathTraverser.java
@@ -184,33 +184,40 @@ public class ShortestPathTraverser extends HugeTraverser {
while (this.pathResults.hasNextKey()) {
Id source = this.pathResults.nextKey();
- Iterator<Edge> edges = edgesOfVertex(source, this.direction,
- this.labels, degree);
- edges = skipSuperNodeIfNeeded(edges, this.degree,
- this.skipDegree);
-
- this.vertexCount += 1L;
-
- while (edges.hasNext()) {
- HugeEdge edge = (HugeEdge) edges.next();
- Id target = edge.id().otherVertexId();
-
- this.edgeResults.addEdge(source, target, edge);
-
- PathSet paths = this.pathResults.findPath(target,
- t ->
!this.superNode(t,
-
this.direction),
- all, false);
-
- if (paths.isEmpty()) {
- continue;
- }
- results.addAll(paths);
- if (!all) {
- return paths;
+ Iterator<Edge> sourceEdges = edgesOfVertex(
+ source, this.direction, this.labels, degree);
+ Throwable failure = null;
+ try {
+ Iterator<Edge> edges = skipSuperNodeIfNeeded(
+ sourceEdges, this.degree, this.skipDegree);
+
+ this.vertexCount += 1L;
+
+ while (edges.hasNext()) {
+ HugeEdge edge = (HugeEdge) edges.next();
+ Id target = edge.id().otherVertexId();
+
+ this.edgeResults.addEdge(source, target, edge);
+
+ PathSet paths = this.pathResults.findPath(
+ target,
+ t -> !this.superNode(t, this.direction),
+ all, false);
+
+ if (paths.isEmpty()) {
+ continue;
+ }
+ results.addAll(paths);
+ if (!all) {
+ return paths;
+ }
}
+ } catch (RuntimeException | Error e) {
+ failure = e;
+ throw e;
+ } finally {
+ closeIterator(sourceEdges, failure);
}
-
}
this.pathResults.finishOneLayer();
@@ -231,30 +238,38 @@ public class ShortestPathTraverser extends HugeTraverser {
while (this.pathResults.hasNextKey()) {
Id source = this.pathResults.nextKey();
- Iterator<Edge> edges = edgesOfVertex(source, opposite,
- this.labels, degree);
- edges = skipSuperNodeIfNeeded(edges, this.degree,
- this.skipDegree);
+ Iterator<Edge> sourceEdges = edgesOfVertex(
+ source, opposite, this.labels, degree);
+ Throwable failure = null;
+ try {
+ Iterator<Edge> edges = skipSuperNodeIfNeeded(
+ sourceEdges, this.degree, this.skipDegree);
- this.vertexCount += 1L;
+ this.vertexCount += 1L;
- while (edges.hasNext()) {
- HugeEdge edge = (HugeEdge) edges.next();
- Id target = edge.id().otherVertexId();
+ while (edges.hasNext()) {
+ HugeEdge edge = (HugeEdge) edges.next();
+ Id target = edge.id().otherVertexId();
- this.edgeResults.addEdge(source, target, edge);
+ this.edgeResults.addEdge(source, target, edge);
- PathSet paths = this.pathResults.findPath(target,
- t ->
!this.superNode(t, opposite),
- all, false);
+ PathSet paths = this.pathResults.findPath(
+ target, t -> !this.superNode(t, opposite),
+ all, false);
- if (paths.isEmpty()) {
- continue;
- }
- results.addAll(paths);
- if (!all) {
- return results;
+ if (paths.isEmpty()) {
+ continue;
+ }
+ results.addAll(paths);
+ if (!all) {
+ return results;
+ }
}
+ } catch (RuntimeException | Error e) {
+ failure = e;
+ throw e;
+ } finally {
+ closeIterator(sourceEdges, failure);
}
}
@@ -270,7 +285,15 @@ public class ShortestPathTraverser extends HugeTraverser {
}
Iterator<Edge> edges = edgesOfVertex(vertex, direction,
this.labels, this.skipDegree);
- return IteratorUtils.count(edges) >= this.skipDegree;
+ Throwable failure = null;
+ try {
+ return IteratorUtils.count(edges) >= this.skipDegree;
+ } catch (RuntimeException | Error e) {
+ failure = e;
+ throw e;
+ } finally {
+ closeIterator(edges, failure);
+ }
}
private long accessed() {
diff --git
a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java
index 1733680e3..91fae045e 100644
---
a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java
+++
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java
@@ -79,6 +79,7 @@ import
org.apache.hugegraph.unit.serializer.TableBackendEntryTest;
import org.apache.hugegraph.unit.serializer.TextBackendEntryTest;
import org.apache.hugegraph.unit.serializer.TextSerializerTest;
import org.apache.hugegraph.unit.store.RamIntObjectMapTest;
+import org.apache.hugegraph.unit.traversal.ShortestPathTraverserTest;
import org.apache.hugegraph.unit.util.CompressUtilTest;
import org.apache.hugegraph.unit.util.JsonUtilTest;
import org.apache.hugegraph.unit.util.RateLimiterTest;
@@ -157,6 +158,7 @@ import org.junit.runners.Suite;
RoleElectionStateMachineTest.class,
HugeGraphAuthProxyTest.class,
SchemaElementTest.class,
+ ShortestPathTraverserTest.class,
/* cmd */
InitStoreConfigTest.class,
diff --git
a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/traversal/ShortestPathTraverserTest.java
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/traversal/ShortestPathTraverserTest.java
new file mode 100644
index 000000000..e3630a110
--- /dev/null
+++
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/traversal/ShortestPathTraverserTest.java
@@ -0,0 +1,429 @@
+/*
+ * 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.unit.traversal;
+
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.backend.id.EdgeId;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.config.CoreOptions;
+import org.apache.hugegraph.structure.HugeEdge;
+import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.traversal.algorithm.HugeTraverser;
+import org.apache.hugegraph.traversal.algorithm.HugeTraverser.Path;
+import org.apache.hugegraph.traversal.algorithm.HugeTraverser.PathSet;
+import org.apache.hugegraph.traversal.algorithm.ShortestPathTraverser;
+import org.apache.hugegraph.type.define.CollectionType;
+import org.apache.hugegraph.type.define.Directions;
+import org.apache.hugegraph.unit.BaseUnitTest;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class ShortestPathTraverserTest extends BaseUnitTest {
+
+ @Test
+ public void testCloseEdgesWhenPathFoundForward() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ TrackingIterator edges = edges(edgeTo(target));
+ TestTraverser traverser = new TestTraverser(edges);
+
+ Path path = shortestPath(traverser, source, target, 1, 0L);
+
+ Assert.assertEquals(Arrays.asList(source, target), path.vertices());
+ Assert.assertTrue(edges.closed());
+ }
+
+ @Test
+ public void testCloseEdgesWhenPathFoundBackward() {
+ Id source = IdGenerator.of(1L);
+ Id middle = IdGenerator.of(2L);
+ Id target = IdGenerator.of(3L);
+ TrackingIterator forwardEdges = edges(edgeTo(middle));
+ TrackingIterator backwardEdges = edges(edgeTo(middle));
+ TestTraverser traverser = new TestTraverser(forwardEdges,
+ backwardEdges);
+
+ Path path = shortestPath(traverser, source, target, 2, 0L);
+
+ Assert.assertEquals(Arrays.asList(source, middle, target),
+ path.vertices());
+ Assert.assertTrue(forwardEdges.closed());
+ Assert.assertTrue(backwardEdges.closed());
+ }
+
+ @Test
+ public void testCloseEdgesWhenCheckingSuperNode() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ TrackingIterator sourceEdges = edges(edgeTo(target));
+ TrackingIterator targetEdges = edges();
+ TestTraverser traverser = new TestTraverser(sourceEdges, targetEdges);
+
+ Path path = shortestPath(traverser, source, target, 1, 2L);
+
+ Assert.assertEquals(Arrays.asList(source, target), path.vertices());
+ Assert.assertTrue(sourceEdges.closed());
+ Assert.assertTrue(targetEdges.closed());
+ }
+
+ @Test
+ public void testCloseEdgesWhenSkipDegreeReached() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ Id other = IdGenerator.of(3L);
+ TrackingIterator sourceEdges = edges(edgeTo(target), edgeTo(other));
+ TestTraverser traverser = new TestTraverser(sourceEdges);
+
+ Path path = shortestPath(traverser, source, target, 1, 2L);
+
+ Assert.assertTrue(path.vertices().isEmpty());
+ Assert.assertTrue(sourceEdges.closed());
+ }
+
+ @Test
+ public void testCloseEdgesWhenTargetIsSuperNode() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ Id other = IdGenerator.of(3L);
+ TrackingIterator sourceEdges = edges(edgeTo(target));
+ TrackingIterator targetEdges = edges(edgeTo(source), edgeTo(other));
+ TestTraverser traverser = new TestTraverser(sourceEdges, targetEdges);
+
+ Path path = shortestPath(traverser, source, target, 1, 2L);
+
+ Assert.assertTrue(path.vertices().isEmpty());
+ Assert.assertTrue(sourceEdges.closed());
+ Assert.assertTrue(targetEdges.closed());
+ }
+
+ @Test
+ public void testCloseEdgesThroughLabelAndLimitWrappers() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ TrackingIterator backendEdges = edges(edgeTo(target));
+ TestTraverser traverser = new TestTraverser(backendEdges);
+
+ Path path = traverser.shortestPath(
+ source, target, Directions.OUT,
+ Collections.singletonList("link"), 1, 1L, 0L, 100L);
+
+ Assert.assertEquals(Arrays.asList(source, target), path.vertices());
+ Assert.assertTrue(backendEdges.closed());
+ }
+
+ @Test
+ public void testCloseAllEdgesThroughWrappersWhenCloseFails() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ RuntimeException closeFailure = new IllegalStateException("close");
+ TrackingIterator first = new TrackingIterator(
+
Arrays.<Edge>asList(edgeTo(target)).iterator(),
+ null,
+ closeFailure);
+ TrackingIterator second = edges();
+ TestTraverser traverser = new TestTraverser(first, second);
+
+ Throwable actual = Assert.assertThrows(
+ RuntimeException.class,
+ () -> traverser.shortestPath(
+ source, target, Directions.OUT,
+ Arrays.asList("first", "second"),
+ 1, 1L, 0L, 100L));
+
+ Assert.assertSame(closeFailure, actual.getCause());
+ Assert.assertTrue(first.closed());
+ Assert.assertTrue(second.closed());
+ }
+
+ @Test
+ public void testCloseEdgesForAllShortestPaths() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ TrackingIterator sourceEdges = edges(edgeTo(target));
+ TestTraverser traverser = new TestTraverser(sourceEdges);
+
+ PathSet paths = traverser.allShortestPaths(
+ source, target, Directions.OUT,
+ Collections.emptyList(), 1, 1L, 0L, 100L);
+
+ Assert.assertEquals(1, paths.size());
+ Assert.assertEquals(Arrays.asList(source, target),
+ paths.iterator().next().vertices());
+ Assert.assertTrue(sourceEdges.closed());
+ }
+
+ @Test
+ public void testCloseAllOpenedEdgesWhenMapLabelQueryFails() {
+ Id source = IdGenerator.of(1L);
+ RuntimeException closeFailure = new IllegalStateException("close");
+ RuntimeException queryFailure = new IllegalArgumentException("query");
+ TrackingIterator first = throwingCloseEdges(closeFailure);
+ TrackingIterator second = edges();
+ TestTraverser traverser = new TestTraverser(first, second,
+ queryFailure);
+ Map<Id, String> labels = new LinkedHashMap<>();
+ labels.put(IdGenerator.of(11L), "first");
+ labels.put(IdGenerator.of(12L), "second");
+ labels.put(IdGenerator.of(13L), "third");
+
+ Throwable actual = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> traverser.queryEdges(
+ source, labels, HugeTraverser.NO_LIMIT));
+
+ Assert.assertSame(queryFailure, actual);
+ Assert.assertTrue(first.closed());
+ Assert.assertTrue(second.closed());
+ assertSuppressedCloseFailure(actual, closeFailure);
+ }
+
+ @Test
+ public void testCloseOpenedEdgesWhenListLabelQueryFails() {
+ Id source = IdGenerator.of(1L);
+ RuntimeException queryFailure = new IllegalArgumentException("query");
+ TrackingIterator first = edges();
+ TestTraverser traverser = new TestTraverser(first, queryFailure);
+ List<Id> labels = Arrays.asList(IdGenerator.of(11L),
+ IdGenerator.of(12L));
+
+ Throwable actual = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> traverser.queryEdges(
+ source, labels, HugeTraverser.NO_LIMIT));
+
+ Assert.assertSame(queryFailure, actual);
+ Assert.assertTrue(first.closed());
+ }
+
+ @Test
+ public void testPreserveTraversalFailureWhenClosingEdgesFails() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ RuntimeException traversalFailure =
+ new IllegalArgumentException("hasNext");
+ RuntimeException closeFailure = new IllegalStateException("close");
+ TrackingIterator sourceEdges = throwingEdges(traversalFailure,
+ closeFailure);
+ TestTraverser traverser = new TestTraverser(sourceEdges);
+
+ Throwable actual = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> shortestPath(traverser, source, target,
+ 1, 0L));
+
+ Assert.assertSame(traversalFailure, actual);
+ Assert.assertTrue(sourceEdges.closed());
+ assertSuppressedCloseFailure(actual, closeFailure);
+ }
+
+ @Test
+ public void testPreserveBackwardFailureWhenClosingEdgesFails() {
+ Id source = IdGenerator.of(1L);
+ Id middle = IdGenerator.of(2L);
+ Id target = IdGenerator.of(3L);
+ RuntimeException traversalFailure =
+ new IllegalArgumentException("hasNext");
+ RuntimeException closeFailure = new IllegalStateException("close");
+ TrackingIterator forwardEdges = edges(edgeTo(middle));
+ TrackingIterator backwardEdges = throwingEdges(traversalFailure,
+ closeFailure);
+ TestTraverser traverser = new TestTraverser(forwardEdges,
+ backwardEdges);
+
+ Throwable actual = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> shortestPath(traverser, source, target,
+ 2, 0L));
+
+ Assert.assertSame(traversalFailure, actual);
+ Assert.assertTrue(forwardEdges.closed());
+ Assert.assertTrue(backwardEdges.closed());
+ assertSuppressedCloseFailure(actual, closeFailure);
+ }
+
+ @Test
+ public void testPreserveSuperNodeFailureWhenClosingEdgesFails() {
+ Id source = IdGenerator.of(1L);
+ Id target = IdGenerator.of(2L);
+ RuntimeException traversalFailure =
+ new IllegalArgumentException("hasNext");
+ RuntimeException closeFailure = new IllegalStateException("close");
+ TrackingIterator sourceEdges = edges(edgeTo(target));
+ TrackingIterator targetEdges = throwingEdges(traversalFailure,
+ closeFailure);
+ TestTraverser traverser = new TestTraverser(sourceEdges, targetEdges);
+
+ Throwable actual = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> shortestPath(traverser, source, target,
+ 1, 2L));
+
+ Assert.assertSame(traversalFailure, actual);
+ Assert.assertTrue(sourceEdges.closed());
+ Assert.assertTrue(targetEdges.closed());
+ assertSuppressedCloseFailure(actual, closeFailure);
+ }
+
+ private static Path shortestPath(TestTraverser traverser, Id source,
+ Id target, int depth, long skipDegree) {
+ return traverser.shortestPath(source, target, Directions.OUT,
+ Collections.emptyList(), depth, 1L,
+ skipDegree, 100L);
+ }
+
+ private static HugeEdge edgeTo(Id target) {
+ HugeEdge edge = Mockito.mock(HugeEdge.class);
+ EdgeId edgeId = Mockito.mock(EdgeId.class);
+ Mockito.when(edge.id()).thenReturn(edgeId);
+ Mockito.when(edgeId.otherVertexId()).thenReturn(target);
+ return edge;
+ }
+
+ private static TrackingIterator edges(Edge... edges) {
+ return new TrackingIterator(Arrays.asList(edges).iterator());
+ }
+
+ private static TrackingIterator throwingCloseEdges(
+ RuntimeException closeFailure) {
+ return new TrackingIterator(Collections.emptyIterator(), null,
+ closeFailure);
+ }
+
+ private static TrackingIterator throwingEdges(
+ RuntimeException traversalFailure,
+ RuntimeException closeFailure) {
+ return new TrackingIterator(Collections.emptyIterator(),
+ traversalFailure, closeFailure);
+ }
+
+ private static void assertSuppressedCloseFailure(
+ Throwable failure, RuntimeException closeFailure) {
+ Assert.assertEquals(1, failure.getSuppressed().length);
+ Assert.assertSame(closeFailure, failure.getSuppressed()[0].getCause());
+ }
+
+ private static HugeGraph mockGraph() {
+ HugeGraph graph = Mockito.mock(HugeGraph.class);
+ Mockito.when(graph.option(CoreOptions.OLTP_COLLECTION_TYPE))
+ .thenReturn(CollectionType.JCF);
+ return graph;
+ }
+
+ private static class TestTraverser extends ShortestPathTraverser {
+
+ private final Deque<Object> edges;
+
+ public TestTraverser(Object... edges) {
+ super(mockGraph());
+ this.edges = new ArrayDeque<>(Arrays.asList(edges));
+ }
+
+ @Override
+ protected void checkVertexExist(Id vertexId, String name) {
+ // Pass: iterator lifecycle is isolated from graph lookup.
+ }
+
+ @Override
+ protected Id getEdgeLabelIdOrNull(Object label) {
+ return label == null ? null : IdGenerator.of(label.toString());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected Iterator<Edge> edgesOfVertex(Id source, Directions dir,
+ Id label,
+ long limit) {
+ Object result = this.edges.removeFirst();
+ if (result instanceof RuntimeException) {
+ throw (RuntimeException) result;
+ }
+ if (result instanceof Error) {
+ throw (Error) result;
+ }
+ return (Iterator<Edge>) result;
+ }
+
+ public Iterator<Edge> queryEdges(Id source, Map<Id, String> labels,
+ long limit) {
+ return super.edgesOfVertex(source, Directions.OUT, labels, limit);
+ }
+
+ public Iterator<Edge> queryEdges(Id source, List<Id> labels,
+ long limit) {
+ return super.edgesOfVertex(source, Directions.OUT, labels, limit);
+ }
+ }
+
+ private static class TrackingIterator implements Iterator<Edge>,
+ AutoCloseable {
+
+ private final Iterator<Edge> edges;
+ private final RuntimeException traversalFailure;
+ private final RuntimeException closeFailure;
+ private boolean closed;
+
+ public TrackingIterator(Iterator<Edge> edges) {
+ this(edges, null, null);
+ }
+
+ public TrackingIterator(Iterator<Edge> edges,
+ RuntimeException traversalFailure,
+ RuntimeException closeFailure) {
+ this.edges = edges;
+ this.traversalFailure = traversalFailure;
+ this.closeFailure = closeFailure;
+ this.closed = false;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (this.traversalFailure != null) {
+ throw this.traversalFailure;
+ }
+ return this.edges.hasNext();
+ }
+
+ @Override
+ public Edge next() {
+ return this.edges.next();
+ }
+
+ @Override
+ public void close() {
+ this.closed = true;
+ if (this.closeFailure != null) {
+ throw this.closeFailure;
+ }
+ }
+
+ public boolean closed() {
+ return this.closed;
+ }
+ }
+}
