Class PipelineGraphFactory

java.lang.Object

ai.pipestream.module.pipelineprobe.e2e.PipelineGraphFactory

@ApplicationScoped
public class PipelineGraphFactory
extends Object

Builds a PipelineGraph from a RunPipelineCrawlRequest's source / steps / observe / transport selection. Pure factory — no I/O, no engine calls, no test orchestration. Easy to unit-test.

Responsibilities:

• Cascade upstream dependencies: SEMANTIC_GRAPH implies EMBEDDER, EMBEDDER implies CHUNKER.
• Drop PARSER when source is JDBC (JDBC reads structured rows; nothing to parse).
• Add a sidecar fan-out tap on every effective step. Taps for steps in observe_steps carry the TestProcessor.OBSERVE_RICH_PARAM flag so the TestProcessor feeds the RichObservationCollector in addition to the always-on TransportTestCounter.
• Always append a testing-sidecar terminal after the last real main-flow step. Per-step taps remain sibling fan-outs for observation.
• Wire main-flow edges. transport_type on proto is legacy (engine uses Kafka pointers + claim-check only; TODO stop setting TransportType).
• Tap fan-out edges to testing-sidecar for per-step observation.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	PipelineGraphFactory.Built	Result of build(String, String, PipelineSource, Set, Set, Set, List, int, Struct, Struct, Struct, Struct, Struct, String) — graph + names callers need to dispatch into it.

Constructor Summary

Constructors

Constructor	Description
PipelineGraphFactory()	Creates the pipeline-graph factory.

Method Summary

Modifier and Type	Method	Description
PipelineGraphFactory.Built	build(String graphId, String clusterId, ai.pipestream.testing.harness.v1.PipelineSource source, Set<ai.pipestream.testing.harness.v1.PipelineStep> requestedSteps, Set<ai.pipestream.testing.harness.v1.PipelineStep> observedSteps, Set<ai.pipestream.testing.harness.v1.PipelineStep> persistedSteps, List<ai.pipestream.testing.harness.v1.E2ETransportType> edgeTransports, int kafkaPartitions, com.google.protobuf.Struct parserCustomConfig, com.google.protobuf.Struct chunkerCustomConfig, com.google.protobuf.Struct embedderCustomConfig, com.google.protobuf.Struct semanticCustomConfig, com.google.protobuf.Struct sinkCustomConfig, String datasourceId)	Build the graph for a request.
void	validateShape(ai.pipestream.testing.harness.v1.PipelineSource source, Set<ai.pipestream.testing.harness.v1.PipelineStep> requestedSteps, List<ai.pipestream.testing.harness.v1.E2ETransportType> edgeTransports)	Validate the request-derived graph shape without building anything.

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

PipelineGraphFactory

public PipelineGraphFactory()

Creates the pipeline-graph factory. The tuning helper collaborator is injected by the container; instances are managed by CDI.

Method Details

validateShape

public void validateShape(ai.pipestream.testing.harness.v1.PipelineSource source,
 Set<ai.pipestream.testing.harness.v1.PipelineStep> requestedSteps,
 List<ai.pipestream.testing.harness.v1.E2ETransportType> edgeTransports)

Validate the request-derived graph shape without building anything. Runs the same step cascade and edge-transport arity checks as build(String, String, PipelineSource, Set, Set, Set, List, int, Struct, Struct, Struct, Struct, Struct, String) and throws the identical IllegalArgumentExceptions. The orchestration driver calls this BEFORE provisioning accounts and datasources, so a request that can never build a graph never mints resources for its own cleanup to race against.

Parameters:

source - the pipeline source (drives the step cascade)

requestedSteps - the requested pipeline steps

edgeTransports - the per-main-flow-edge transport selection to validate

Throws:

IllegalArgumentException - if the edge-transport count does not match the number of main-flow edges the cascade would produce

build

public PipelineGraphFactory.Built build(String graphId,
 String clusterId,
 ai.pipestream.testing.harness.v1.PipelineSource source,
 Set<ai.pipestream.testing.harness.v1.PipelineStep> requestedSteps,
 Set<ai.pipestream.testing.harness.v1.PipelineStep> observedSteps,
 Set<ai.pipestream.testing.harness.v1.PipelineStep> persistedSteps,
 List<ai.pipestream.testing.harness.v1.E2ETransportType> edgeTransports,
 int kafkaPartitions,
 com.google.protobuf.Struct parserCustomConfig,
 com.google.protobuf.Struct chunkerCustomConfig,
 com.google.protobuf.Struct embedderCustomConfig,
 com.google.protobuf.Struct semanticCustomConfig,
 com.google.protobuf.Struct sinkCustomConfig,
 String datasourceId)

Build the graph for a request. Configs that need per-step custom_config are passed in by the caller (avoid coupling the factory to test-state).

Parameters:

graphId - the graph id to assign

clusterId - the cluster id to assign

source - the pipeline source (drives the step cascade)

requestedSteps - the requested pipeline steps (before cascade)

observedSteps - steps whose taps run in rich-observation mode

persistedSteps - steps whose nodes keep their consumed input claim-check body for replay

edgeTransports - per-main-flow-edge transport selection; empty/null fills with Kafka placeholders

kafkaPartitions - kafka-partition count applied to main-flow nodes

parserCustomConfig - custom config for the parser node, or null

chunkerCustomConfig - custom config for the chunker node, or null

embedderCustomConfig - custom config for the embedder node, or null

semanticCustomConfig - custom config for the semantic-graph node, or null

sinkCustomConfig - custom config for the opensearch-sink node, or null

datasourceId - datasource id declared on the virtual intake node

Returns:

the built graph plus the node-ids callers need to dispatch into and wait on it

Throws:

IllegalArgumentException - if the edge-transport count does not match the number of main-flow edges, including the tap-only case where it must be empty