Source Code Reading: PipelineService — DSL to DAG Compilation

“

Series: S9 Source Code Reading · Article 16 | Level: Advanced | Reading Time: 25 min

#TL;DR

PipelineService is the core service in coomia-dip's data pipeline layer (Pipeline & Orchestration Layer, merged into Data Layer), responsible for compiling declarative Pipeline DSL into executable Directed Acyclic Graphs (DAGs) and scheduling their execution. Built on Quarkus 3.x, it exposes CRUD, execution, and monitoring API groups via gRPC. This article dissects the PipelineServiceGrpcImpl adapter pattern, the PipelineExecutionService application service layer's pipeline definition management, the Quarkus CDI proxy and Mutiny Uni scoping pitfall, and the PipelineDefinition Step-DAG compilation model in the Proto contract.

#Table of Contents

1. Overall Architecture: Pipeline & Orchestration Layer's Quarkus gRPC Service
2. Proto Contract: PipelineDefinition and Step-DAG Model
3. PipelineServiceGrpcImpl: Adapter Pattern
4. Quarkus CDI Proxy Scoping Pitfall
5. CreatePipeline: Definition and Validation
6. Pipeline Execution: DAG Scheduling Strategy
7. Lineage Recording: Pipeline and LineageService Integration
8. Error Handling: gRPC Status Mapping
9. Metadata and Version Management
10. Pipeline Monitoring and Observability
11. Key Takeaways

#1. Overall Architecture: Pipeline & Orchestration Layer's Quarkus gRPC Service

The Pipeline service resides in the Data Layer (after Data Layer + Pipeline & Orchestration Layer merge), built on Quarkus 3.x:

data-Layer/src/main/java/com/onto/plane_f/
├── api/grpc/
│   └── PipelineServiceGrpcImpl.java    # gRPC adapter layer
├── service/
│   └── PipelineExecutionService.java    # Application service layer
proto/plane_f/
├── pipeline_engine.proto                # Pipeline core contract
└── lineage.proto                        # Lineage tracking contract

Technology choice: Pipeline service uses Quarkus rather than Spring Boot because the Data Layer needs minimal startup time (GraalVM native image support) and smaller memory footprint. Quarkus's Mutiny reactive model also naturally fits Pipeline's async execution needs.

#2. Proto Contract: PipelineDefinition and Step-DAG Model

The Pipeline Proto definition follows a Step-DAG model -- each Pipeline consists of Steps with dependency relationships. The depends_on field compiles the Step list into a DAG: a Step is only scheduled for execution when all its dependencies complete. The compiler detects cyclic dependencies and rejects Pipeline definitions containing cycles.

Step types follow the ETL paradigm: EXTRACT (extract from external sources), TRANSFORM (data transformation and cleaning), LOAD (load to target storage), and CUSTOM (user-defined steps referencing FunctionRuntime functions).

#3. PipelineServiceGrpcImpl: Adapter Pattern

PipelineServiceGrpcImpl is a typical thin adapter containing no business logic:

@GrpcService
public class PipelineServiceGrpcImpl implements PipelineService {
    @Inject
    PipelineExecutionService service;

    @Override
    public Uni<PipelineResponse> createPipeline(CreatePipelineRequest request) {
        try {
            PipelineDefinition inputDef = request.getPipeline();
            if (inputDef == null) {
                return Uni.createFrom().failure(Status.INVALID_ARGUMENT
                    .withDescription("pipeline definition is required")
                    .asRuntimeException());
            }

            PipelineDefinition storedDef = service.createPipeline(inputDef);

            PipelineResponse response = PipelineResponse.newBuilder()
                .setPipeline(storedDef)
                .setCreatedAt(createdAt)
                .setUpdatedAt(updatedAt)
                .setVersionId(versionId)
                .build();

            return Uni.createFrom().item(response);
        } catch (PipelineAlreadyExistsException e) {
            return Uni.createFrom().failure(Status.ALREADY_EXISTS
                .withDescription(e.getMessage()).asRuntimeException());
        }
    }
}

Key pattern: All business logic calls execute synchronously in the method body, then wrap the result in Uni.createFrom().item(value). This is a deliberate design decision explained next.

#4. Quarkus CDI Proxy Scoping Pitfall

A critical comment in the code reveals a Quarkus CDI pitfall:

/**
 * IMPORTANT: All service calls are executed synchronously in the method body
 * (NOT inside Uni lambdas) to avoid Quarkus CDI proxy scoping issues where
 * Uni.createFrom().item(Supplier) lambdas resolve to different bean instances,
 * causing the in-memory pipeline store to appear empty.
 */

When using Uni.createFrom().item(() -> service.createPipeline(inputDef)), the service reference inside the lambda may resolve to a different CDI proxy instance -- particularly when PipelineExecutionService uses @ApplicationScoped. The lambda may execute on a different thread, seeing an "empty" memory store.

Solution: Call service.createPipeline(inputDef) synchronously first to get the result, then pass the result (not a Supplier) to Uni.createFrom().item(value). This is a classic Quarkus + CDI + Mutiny integration trap that the coomia-dip team discovered and documented for future maintainers.

#5. CreatePipeline: Definition and Validation

The Pipeline creation flow includes multi-layer validation with three exception types forming clear error classification: PipelineAlreadyExistsException maps to ALREADY_EXISTS (name uniqueness conflict), IllegalArgumentException maps to INVALID_ARGUMENT (input validation failure such as DAG containing cycles), and generic Exception maps to INTERNAL.

Metadata null-safety: When building responses, every Metadata field uses null-safe handling:

Timestamp createdAt = storedDef.hasMetadata()
    ? storedDef.getMetadata().getCreatedAt()
    : Timestamp.getDefaultInstance();

Using hasMetadata() check instead of directly calling getMetadata() avoids NullPointerException.

#6. Pipeline Execution: DAG Scheduling Strategy

Pipeline execution uses topological sorting to determine Step execution order:

1. Topological sort: Parse depends_on to build DAG, perform topological sort
2. Parallel scheduling: Steps at the same topological level (no dependencies between them) can execute in parallel
3. Failure handling: When a Step fails, all its downstream Steps are automatically marked SKIPPED
4. Retry mechanism: Supports Step-level retry policies

DAG compilation occurs at Pipeline creation time -- PipelineExecutionService.createPipeline() validates DAG legality (acyclic, all referenced step_ids exist). At runtime, scheduling simply follows the topological order.

#7. Lineage Recording: Pipeline and LineageService Integration

Pipeline execution automatically records lineage information to Pipeline & Orchestration Layer's LineageService:

message RecordLineageRequest {
  ExecutionContext context = 1;
  repeated DatasetRef inputs = 2;
  repeated DatasetRef outputs = 3;
  repeated FieldMapping field_mappings = 4;
  map<string, string> metadata = 5;
}

Each Step completion records its input and output dataset references plus field-level mapping relationships. This provides complete chain information for post-hoc data tracing and impact analysis.

FieldMapping supports six transform types: DIRECT (direct copy), AGGREGATE, JOIN, FILTER, UDF (user-defined function), and EXPRESSION.

#8. Error Handling: gRPC Status Mapping

The adapter layer maps business exceptions to standard gRPC Status codes: PipelineAlreadyExistsException to ALREADY_EXISTS (HTTP 409), IllegalArgumentException to INVALID_ARGUMENT (HTTP 400), PipelineNotFoundException to NOT_FOUND (HTTP 404), and generic Exception to INTERNAL (HTTP 500).

#9. Metadata and Version Management

PipelineDefinition's Metadata field carries rich management information including created_by, created_at, version, labels, and annotations. Version number retrieval has priority logic:

String versionId = storedDef.hasMetadata()
    ? storedDef.getMetadata().getVersion()
    : storedDef.getVersion();

Metadata version (server-managed) takes priority, falling back to PipelineDefinition's top-level version field (client-specified).

#10. Pipeline Monitoring and Observability

Pipeline execution provides multi-dimensional observability: Step-level status tracking (PENDING/RUNNING/COMPLETED/FAILED/SKIPPED), execution timeline (started_at/completed_at per Step), resource usage via Quarkus Micrometer integration, and log correlation through execution_id linking all logs for a Pipeline execution.

The PipelineServiceGrpcImpl records structured logs at every RPC entry point for operational visibility.

#11. Key Takeaways

1. Thin adapter pattern: gRPC Impl handles only parameter validation, exception mapping, and Proto conversion; business logic lives entirely in the Service layer.
2. CDI proxy pitfall: Uni lambda interaction with CDI proxies in Quarkus requires careful attention -- call synchronously first, then wrap in Uni.
3. Step-DAG compilation model: depends_on fields compile Step lists into DAGs, validated at creation time and scheduled at runtime.
4. Three-layer exception classification: ALREADY_EXISTS / INVALID_ARGUMENT / INTERNAL covers all error scenarios.
5. Metadata null-safety: hasMetadata() + default value fallback ensures no crashes even with missing Metadata.
6. Automatic lineage recording: Pipeline-LineageService integration makes data tracing a zero-cost operation.

#Next Article

S9-17: AuditService -- Cross-Process Audit Kafka Consumer, where we dive into Metadata & Governance Layer's audit service to understand decision tracing and compliance audit implementation.

Tags: #coomia-dip #source-code-reading #pipeline-service #dsl-to-dag #quarkus #grpc #Layer-f