Class PipelineEventsConsumer

java.lang.Object
ai.pipestream.module.pipelineprobe.pipelinecrawl.PipelineEventsConsumer
@ApplicationScoped public class PipelineEventsConsumer extends Object
Live observation feed for the Pipeline-tab panels. Consumes the engine's pipeline-events audit stream — the SAME StepExecutionRecord stream opensearch-manager indexes — and folds each record into RichObservationCollector, keyed by crawl_id.

This is the one Kafka subscriber in the sidecar. It sees every crawl's records platform-wide, so it folds ONLY records whose crawl_id has an open run (ActiveCrawlRegistry); everything else is dropped, keeping the collector bounded to live runs (and making auto.offset.reset=earliest harmless — replayed history has no active runs).

Serialization is fully handled by the quarkus-apicurio-registry-protobuf extension: it auto-configures the channel (connector, topic pipeline-events, UUID key deserializer, protobuf value deserializer, offset, ignore-failure). The record key is the engine's stream UUID; we don't use it.

One (crawl, node, doc) resolves to exactly one outcome — the engine emits a rich success record AND a separate NO_OP/REJECT disposition record for the same doc, so RichObservationCollector classifies exclusively with precedence reject > no-op > processed (order-independent).

  • Constructor Details

    • PipelineEventsConsumer

      public PipelineEventsConsumer()
      Default constructor for CDI.

  • Method Details

    • onStepRecord

      public void onStepRecord(io.smallrye.reactive.messaging.kafka.Record<UUID, ai.pipestream.data.v1.StepExecutionRecord> record)
      Fold one audit record into the observation collector for its crawl.
      Parameters:
      record - Kafka record: UUID stream key + the StepExecutionRecord value