This is an automated email from the ASF dual-hosted git repository. kenhuuu pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit d3b8ca636412efb51e77d179c95a7a1eb71bf74c Author: Ken Hu <[email protected]> AuthorDate: Mon Jun 8 16:13:12 2026 -0700 Fix transaction executor thread leak on commit and rollback CTR Assisted-by: Claude Code:claude-opus-4-6 --- .../gremlin/server/channel/HttpChannelizer.java | 1 - .../server/handler/HttpGremlinEndpointHandler.java | 4 +- .../TransactionManager.java | 11 +++-- .../UnmanagedTransaction.java | 2 +- .../gremlin/server/util/ServerGremlinExecutor.java | 2 +- .../GremlinServerHttpTransactionIntegrateTest.java | 52 +++++++++++++++++++++- 6 files changed, 61 insertions(+), 11 deletions(-) diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/HttpChannelizer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/HttpChannelizer.java index c25bc3bfcc..9d599199d5 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/HttpChannelizer.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/HttpChannelizer.java @@ -36,7 +36,6 @@ import org.apache.tinkerpop.gremlin.server.handler.HttpRequestIdHandler; import org.apache.tinkerpop.gremlin.server.handler.HttpRequestMessageDecoder; import org.apache.tinkerpop.gremlin.server.handler.HttpUserAgentHandler; import org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler; -import org.apache.tinkerpop.gremlin.server.handler.TransactionManager; import io.netty.channel.ChannelPipeline; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java index b520464d68..8bc6e79c3e 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java @@ -50,6 +50,8 @@ import org.apache.tinkerpop.gremlin.server.GremlinServer; import org.apache.tinkerpop.gremlin.server.ProcessingException; import org.apache.tinkerpop.gremlin.server.Settings; import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser; +import org.apache.tinkerpop.gremlin.server.transaction.TransactionManager; +import org.apache.tinkerpop.gremlin.server.transaction.UnmanagedTransaction; import org.apache.tinkerpop.gremlin.server.util.GremlinError; import org.apache.tinkerpop.gremlin.server.util.MetricManager; import org.apache.tinkerpop.gremlin.server.util.TraverserIterator; @@ -541,7 +543,7 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ final Pair<String, MessageSerializer<?>> serializer) throws Exception { final Graph graph = graphManager.getTraversalSource(ctx.getRequestMessage().getField(Tokens.ARGS_G)).getGraph(); graphOp.accept(graph.tx()); - transactionManager.destroy(transactionId); + transactionManager.get(transactionId).ifPresent(tx -> tx.close(true)); final ByteBuf chunk = makeChunk(ctx, serializer.getValue1(), List.of(Map.of(Tokens.ARGS_TRANSACTION_ID, transactionId)), false, false); ctx.getChannelHandlerContext().writeAndFlush(new DefaultHttpContent(chunk)); } diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/TransactionManager.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/transaction/TransactionManager.java similarity index 95% rename from gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/TransactionManager.java rename to gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/transaction/TransactionManager.java index 3408a581b9..6c4e8b0f76 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/TransactionManager.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/transaction/TransactionManager.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.tinkerpop.gremlin.server.handler; +package org.apache.tinkerpop.gremlin.server.transaction; import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource; import org.apache.tinkerpop.gremlin.server.GraphManager; @@ -99,12 +99,11 @@ public class TransactionManager { } /** - * Removes a transaction from the active transactions map. Called when a transaction is - * committed, rolled back, or otherwise closed. - * - * @param id The transaction ID to remove + * Removes a transaction from the active transactions map. Package-private so that only + * {@link UnmanagedTransaction#close(boolean)} can call it — external callers must go + * through {@code close()} to ensure the executor is shut down. */ - public void destroy(final String id) { + void destroy(final String id) { transactions.remove(id); } diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/UnmanagedTransaction.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/transaction/UnmanagedTransaction.java similarity index 99% rename from gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/UnmanagedTransaction.java rename to gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/transaction/UnmanagedTransaction.java index ad8d0e3baf..0b98b9b846 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/UnmanagedTransaction.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/transaction/UnmanagedTransaction.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.tinkerpop.gremlin.server.handler; +package org.apache.tinkerpop.gremlin.server.transaction; import org.apache.tinkerpop.gremlin.structure.Graph; import org.slf4j.Logger; diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/ServerGremlinExecutor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/ServerGremlinExecutor.java index fa569d6aeb..570cf8d7da 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/ServerGremlinExecutor.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/ServerGremlinExecutor.java @@ -29,7 +29,7 @@ import org.apache.tinkerpop.gremlin.server.Channelizer; import org.apache.tinkerpop.gremlin.server.GraphManager; import org.apache.tinkerpop.gremlin.server.GremlinServer; import org.apache.tinkerpop.gremlin.server.Settings; -import org.apache.tinkerpop.gremlin.server.handler.TransactionManager; +import org.apache.tinkerpop.gremlin.server.transaction.TransactionManager; import org.apache.tinkerpop.gremlin.structure.Graph; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpTransactionIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpTransactionIntegrateTest.java index ce21743c50..8aa919cad8 100644 --- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpTransactionIntegrateTest.java +++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpTransactionIntegrateTest.java @@ -31,7 +31,7 @@ import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; import org.apache.http.util.EntityUtils; import org.apache.tinkerpop.gremlin.server.channel.HttpChannelizer; -import org.apache.tinkerpop.gremlin.server.handler.TransactionManager; +import org.apache.tinkerpop.gremlin.server.transaction.TransactionManager; import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONTokens; import org.apache.tinkerpop.gremlin.util.Tokens; import org.apache.tinkerpop.gremlin.util.message.RequestMessage; @@ -647,6 +647,56 @@ public class GremlinServerHttpTransactionIntegrateTest extends AbstractGremlinSe final byte[] asArray = EntityUtils.toByteArray(httpEntity); return Unpooled.wrappedBuffer(asArray); } + + @Test + public void shouldNotLeakTransactionExecutorThreadOnCommit() throws Exception { + final long txThreadsBefore = Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().startsWith("tx-")).count(); + + for (int i = 0; i < 3; i++) { + final String txId = beginTx(client, GTX); + try (final CloseableHttpResponse r = submitInTx(client, txId, "g.addV('leak_test')", GTX)) { + assertEquals(200, r.getStatusLine().getStatusCode()); + } + try (final CloseableHttpResponse r = commitTx(client, txId, GTX)) { + assertEquals(200, r.getStatusLine().getStatusCode()); + } + } + + // allow time for executor threads to terminate after shutdown + Thread.sleep(500); + + final long txThreadsAfter = Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().startsWith("tx-")).count(); + + assertEquals("Transaction executor threads should be cleaned up after commit", + txThreadsBefore, txThreadsAfter); + } + + @Test + public void shouldNotLeakTransactionExecutorThreadOnRollback() throws Exception { + final long txThreadsBefore = Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().startsWith("tx-")).count(); + + for (int i = 0; i < 3; i++) { + final String txId = beginTx(client, GTX); + try (final CloseableHttpResponse r = submitInTx(client, txId, "g.addV('leak_test')", GTX)) { + assertEquals(200, r.getStatusLine().getStatusCode()); + } + try (final CloseableHttpResponse r = rollbackTx(client, txId, GTX)) { + assertEquals(200, r.getStatusLine().getStatusCode()); + } + } + + // allow time for executor threads to terminate after shutdown + Thread.sleep(500); + + final long txThreadsAfter = Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().startsWith("tx-")).count(); + + assertEquals("Transaction executor threads should be cleaned up after rollback", + txThreadsBefore, txThreadsAfter); + } @Test public void shouldRoundTripTransactionIdWithGraphSON() throws Exception { final GraphSONMessageSerializerV4 serializer = new GraphSONMessageSerializerV4();
