Pipeline Notifications

The pipeline publishes event notifications after each destination loader completes. Notifications are sent to an ActiveMQ Virtual Topic and can be consumed by any number of independent subscribers.

How It Works

The pipeline uses ActiveMQ’s Virtual Destinations pattern:

Each destination loader (PostgreSQL, MongoDB, Snowflake, Databricks, Kafka, ActiveMQ, Object Store, REST Endpoint, Qdrant, Weaviate, Milvus, Chroma, pgvector) publishes a notification to VirtualTopic.{environment}-pipeline-notification
Consumers subscribe by creating a queue named Consumer.{name}.VirtualTopic.{environment}-pipeline-notification
ActiveMQ automatically copies topic messages to all matching consumer queues
Each consumer gets its own independent queue with its own message cursor

No subscription API is needed — consumers simply connect to their named queue.

Consuming Notifications

Queue Naming Convention

Consumer.{consumer-name}.VirtualTopic.{environment}-pipeline-notification

For example, with environment oss:

Consumer.analytics.VirtualTopic.oss-pipeline-notification
Consumer.audit-log.VirtualTopic.oss-pipeline-notification
Consumer.dashboard.VirtualTopic.oss-pipeline-notification

Each consumer queue receives a copy of every notification independently.

Quick Start: Python Consumer (Minimal)

import stomp
import json

class NotificationListener(stomp.ConnectionListener):
    def on_message(self, frame):
        notification = json.loads(frame.body)
        print(f"Pipeline: {notification['pipeline']}, Destination: {notification['destination']}")

conn = stomp.Connection([('localhost', 61613)])
conn.set_listener('', NotificationListener())
conn.connect('admin', 'admin', wait=True)
conn.subscribe(
    destination='/queue/Consumer.myapp.VirtualTopic.oss-pipeline-notification',
    id=1,
    ack='auto'
)

Note: This is a minimal snippet. For a production-ready consumer with automatic reconnect, heartbeats, graceful shutdown, and configurable virtual topic support, see Topic Subscriber.

Example: Java/JMS Consumer

ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = factory.createConnection("admin", "admin");
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("Consumer.myapp.VirtualTopic.oss-pipeline-notification");
MessageConsumer consumer = session.createConsumer(queue);
consumer.setMessageListener(message -> {
    TextMessage text = (TextMessage) message;
    System.out.println("Notification: " + text.getText());
});

Message Filtering

Notifications include JMS message properties that can be used with JMS selectors for filtering:

Property	Description
`pipeline`	Pipeline name
`destination`	Destination type (`postgres`, `mongodb`, `snowflake`, `databricks`, `objectStore`, `kafka`, `activemq`, `restEndpoint`, `qdrant`, `weaviate`, `milvus`, `chroma`, `pgvector`)
`schema`	Database schema (database destinations)
`database`	Database name (database destinations)
`table`	Table name (database destinations)
`topic`	Kafka topic (Kafka destinations)
`queueName`	Queue name (ActiveMQ destinations)
`prefixKey`	Object store prefix (object store destinations)

To filter, use a JMS selector when creating the consumer:

MessageConsumer consumer = session.createConsumer(queue, "pipeline = 'sales_data' AND destination = 'postgres'");

Notification Payload

{
  "pipeline": "sales_data",
  "publisherToken": "publisher-123",
  "pipelineToken": "pt-abc12345-...",
  "destination": "postgres",
  "prefixKey": null,
  "objectStoreUrl": null,
  "objectStoreTemporaryUrl": null,
  "schema": "public",
  "database": "analytics",
  "table": "sales",
  "topic": null,
  "queueName": null
}

Fields are populated based on the destination type. Unused fields are null.

Enabling Notifications

Notifications are enabled by default in application.yaml:

sendPipelineNotifications: "true"

Set to "false" to disable all notifications.

Getting Started

Assistant

Catalog

Taps

Ingestion

Destinations

Data Quality

Transformation

AI Features

Configuration

Integrations

Examples

Pipeline Notifications

How It Works

Consuming Notifications

Queue Naming Convention

Quick Start: Python Consumer (Minimal)

Example: Java/JMS Consumer

Message Filtering

Notification Payload

Enabling Notifications

​How It Works

​Consuming Notifications

​Queue Naming Convention

​Quick Start: Python Consumer (Minimal)

​Example: Java/JMS Consumer

​Message Filtering

​Notification Payload

​Enabling Notifications

How It Works

Consuming Notifications

Queue Naming Convention

Quick Start: Python Consumer (Minimal)

Example: Java/JMS Consumer

Message Filtering

Notification Payload

Enabling Notifications