Event Streaming & Kafka (MCP/MCE)
Relevant source files
The following files were used as context for generating this wiki page:
- metadata-jobs/mce-consumer-job/README.md
- metadata-service/configuration/src/main/resources/application.yaml
- metadata-integration/java/acryl-spark-lineage/src/main/java/datahub/spark/converter/SparkStreamingEventToDatahub.java
- metadata-ingestion/KAFKA_CONNECT_LINEAGE.md
- docs-website/static/llms.txt
- metadata-ingestion/developing.md
DataHub utilizes an event-driven architecture centered around Kafka to manage metadata changes. This system relies on two primary message types: Metadata Change Proposals (MCP) and Metadata Change Events (MCE). These events allow DataHub to decouple metadata ingestion from the internal storage and indexing logic, providing a scalable way to handle updates from various sources such as Spark streaming jobs, Kafka Connectors, and direct CLI ingestion.
Sources: docs-website/static/llms.txt, metadata-jobs/mce-consumer-job/README.md
Core Event Types
The architecture transitions from the legacy Metadata Change Event (MCE) format to the modern Metadata Change Proposal (MCP) and Metadata Change Log (MCL) framework.
Metadata Change Proposal (MCP)
MCPs are requests sent by clients (e.g., ingestion sources) to change the metadata graph. They are per-aspect proposals rather than full entity snapshots. MCPs are primarily consumed by the Metadata Service (GMS) or a standalone MCE Consumer.
Sources: metadata-jobs/mce-consumer-job/README.md:14-18
Metadata Change Log (MCL)
Once a change is committed to the storage layer, the system emits an MCL. This acts as a source of truth for downstream consumers, such as search indexers (MAE consumers) or external audit systems.
Sources: metadata-service/configuration/src/main/resources/application.yaml:692-699
Event Topic Mapping
DataHub uses specific Kafka topics for different stages of the metadata lifecycle:
| Logical Identifier | Kafka Topic Name | Purpose |
|---|---|---|
metadataChangeProposal | MetadataChangeProposal_v1 | Inbound requests for metadata changes. |
failedMetadataChangeProposal | FailedMetadataChangeProposal_v1 | Dead-letter queue for failed MCP processing. |
metadataChangeLogVersioned | MetadataChangeLog_Versioned_v1 | Logs of changes for versioned aspects. |
metadataChangeLogTimeseries | MetadataChangeLog_Timeseries_v1 | Logs of changes for timeseries aspects (e.g., stats). |
platformEvent | PlatformEvent_v1 | Internal platform events. |
Sources: metadata-service/configuration/src/main/resources/application.yaml:810-845, metadata-jobs/mce-consumer-job/README.md:14-22
Architecture & Data Flow
The flow of metadata typically begins at an external system and ends in the DataHub storage and search indices via Kafka.
High-Level Ingestion Flow
The diagram shows the ingestion path from an external source through Kafka topics to final storage and indexing.
Sources: metadata-jobs/mce-consumer-job/README.md:7-12, metadata-service/configuration/src/main/resources/application.yaml:810-830
Spark Streaming Metadata Extraction
DataHub provides specialized converters for Spark Streaming. These converters translate StreamingQueryProgress events into MCPs for dataFlow and dataJob entities.
- Logic: Stable identifiers for streaming queries are generated from the query name or sanitized sink descriptions.
- Platform Mapping: Detects platforms like
kafka,delta-lake, andfilefrom streaming namespaces.
Sources: metadata-integration/java/acryl-spark-lineage/src/main/java/datahub/spark/converter/SparkStreamingEventToDatahub.java:41-75
Kafka Connect Lineage Extraction
The Kafka Connect ingestion framework extracts lineage by mapping source tables to Kafka topics (for source connectors) or Kafka topics to target tables (for sink connectors).
Source Connector Mapping Logic
Source connectors map internal database concepts (e.g., schema.users) to Kafka topics. The system applies a "Transform Pipeline" (such as a RegexRouter) to predict the final topic name used in DataHub.
Sources: metadata-ingestion/KAFKA_CONNECT_LINEAGE.md:140-160
Sink Connector Mapping (BigQuery Example)
Sink connectors, like the BigQuery Sink, map Kafka topics to target datasets.
- Priority Resolution:
- Explicit
topic2TableMapin connector config. - Regex-based
topicsToTablespatterns. - Sanitized topic name as the default table name.
Sources: metadata-ingestion/KAFKA_CONNECT_LINEAGE.md:65-85
- Explicit
The sequence shows how DataHub interprets Kafka Connect configurations to establish end-to-end lineage.
Sources: metadata-ingestion/KAFKA_CONNECT_LINEAGE.md:140-190
Configuration & Tuning
Kafka integration is configured primarily through application.yaml.
Consumer & Producer Settings
| Parameter | Default | Description |
|---|---|---|
kafka.producer.compressionType | snappy | Compression algorithm for outbound messages. |
kafka.consumer.maxPartitionFetchBytes | 5242880 | Max bytes consumed per partition. |
kafka.listener.concurrency | 1 | Number of concurrent listener threads. |
kafka.producer.retryCount | 3 | Number of retries for message delivery. |
Sources: metadata-service/configuration/src/main/resources/application.yaml:894-915
Throttling & Backlog Management
DataHub can monitor the MCL backlog to throttle inbound requests if the system is overloaded. This is particularly relevant for versioned and timeseries aspects.
- Thresholds: Configurable thresholds (default 4000) determine when to start throttling.
- Backoff: Uses exponential backoff (initial 100ms, multiplier 10x) up to a maximum interval (default 30s).
Sources: metadata-service/configuration/src/main/resources/application.yaml:1130-1155
The MCE Consumer Job
The MCE Consumer is a Spring-based job (can be standalone or embedded in GMS) that listens to MetadataChangeProposal_v1.
- Parallelization: In standalone mode, processing parallelization is determined by
CPU cores + 1. - Connectivity: It applies changes against the Metadata Service (GMS) storage layer.
- Error Handling: Failed proposals are routed to
FailedMetadataChangeProposal_v1.
Sources: metadata-jobs/mce-consumer-job/README.md:27-32, 70-75
Conclusion
DataHub’s Kafka-based architecture ensures that metadata changes are processed reliably and asynchronously. By standardizing on the MCP/MCL framework, the system supports complex lineage extraction from Kafka Connect and real-time updates from Spark Streaming, while providing robust configuration options for performance tuning and error management.
Sources: metadata-jobs/mce-consumer-job/README.md, metadata-ingestion/developing.md:65-70