Class TestGraphCleanupService

java.lang.Object
ai.pipestream.module.pipelineprobe.cleanup.TestGraphCleanupService
@ApplicationScoped public class TestGraphCleanupService extends Object
Batch cleanup of leftover test graphs. Enumerates graphs by reading Consul node-topic KV keys (each graph with Kafka hops registers slot keys under pipestream/node-topics/{cluster}.{graph}.*), filters by configured test-graph prefixes, and issues PipelineGraphServiceGrpc.PipelineGraphServiceBlockingStub.deleteAllGraphVersions(DeleteAllGraphVersionsRequest) for each.

Scoped to test prefixes so production/user graphs are never touched. Callers can override the prefixes to broaden or narrow the scope.

No Mutiny: a blocking gRPC stub plus the plain Vert.x ConsulClient (from the platform's PipestreamConsulClientProducer), whose async calls are bridged to blocking via ConsulFutures.await(Future). All public methods must therefore be invoked on a virtual thread so the bridge parks without pinning the event loop.

  • Constructor Details

    • TestGraphCleanupService

      public TestGraphCleanupService()
      Default constructor for CDI.

  • Method Details

    • listTestGraphIds

      public List<String> listTestGraphIds()
      Lists every graph ID whose slot keys currently live in Consul that matches a configured test-prefix. Returns distinct graph IDs in discovery order.
      Returns:
      distinct test-prefixed graph IDs in discovery order

    • deleteAllTestGraphs

      public TestGraphCleanupService.CleanupSummary deleteAllTestGraphs()
      Deletes all graph versions for every test-prefixed graph currently registered in Consul. Each DeleteAllGraphVersions call runs sequentially so the engine's per-graph state unwinds cleanly.
      Returns:
      a summary of graph IDs attempted + their delete outcomes

    • releaseAllTestLeases

      public int releaseAllTestLeases()
      Releases every Consul KV lease key under pipestream/node-topics/ whose graph segment matches a configured test prefix (testGraphPrefixesCsv). Belt-and-suspenders for deleteAllTestGraphs() — the engine's DeleteAllGraphVersions should cascade-delete the slot keys as part of its graph teardown, but we have observed leases linger past graph deletion (issue map section E). Always-fire this delete keeps the kafka sidecar from saturating its max-leases ceiling.
      Returns:
      the number of clusters whose lease keys were released; never throws — Consul errors are logged at WARN and swallowed

    • releaseLeasesForCluster

      public void releaseLeasesForCluster(String clusterPrefix)
      Releases every Consul KV lease key under pipestream/node-topics/{clusterPrefix}.* — the lock entries the kafka sidecar uses to acquire ownership of node topics for a graph.

      Without this, every transport test run leaks 2 lease keys (one per node topic) regardless of skip_cleanup; once the sidecar holds pipestream.sidecar.max-leases stale entries, every subsequent Kafka-mode crawl hangs at zero docs because the new graph's topics never get a consumer subscription.

      clusterPrefix is typically the test account id (which doubles as the cluster id for harness-generated graphs). The matching prefix is {nodeTopicsPath}/{clusterPrefix}. — the trailing dot pins the match to a full segment so a longer-prefix accident can't cascade.

      Parameters:
      clusterPrefix - the cluster/account id whose lease keys to drop; blank is a no-op