Metadata Change Events Not Processing

Metadata Change Events (MCEs) or Metadata Change Proposals (MCPs) emitted to DataHub are not being processed or reflected in the UI. This can occur due to Kafka connectivity issues, malformed event payloads, schema validation failures, or processing errors in the DataHub backend.

Common Error Messages:

• Failed to produce to Kafka topic
• Schema validation failed for MCP
• Invalid entity URN format
• Aspect not registered

You Might Be Asking:

• How do I check if my MCPs are being processed?
• Why aren't my metadata changes showing up?
• What's the difference between MCE and MCP?

1. (for DataHub OSS instances) Verify Kafka connectivity and topic health:

# For DataHub open source
# Check Kafka topics
kubectl exec -it kafka-0 -- kafka-topics.sh --list --bootstrap-server localhost:9092

# Verify MetadataChangeProposal_v1 topic exists
# Check for consumer lag
kubectl exec -it kafka-0 -- kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group datahub-mce-consumer --describe

2. Validate MCP format before emitting:

Example for Snowflake. This can be adapted for other data sources/connectors.

from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import DatasetPropertiesClass

# Correct MCP format
emitter = DatahubRestEmitter("http://localhost:8080")

dataset_properties = DatasetPropertiesClass(
    description="Updated description",
    customProperties={"key": "value"}
)

# Emit and check for errors
try:
    emitter.emit_mcp(
        entity_urn="urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table,PROD)",
        aspect_name="datasetProperties",
        aspect=dataset_properties
    )
    print("MCP emitted successfully")
except Exception as e:
    print(f"Error emitting MCP: {e}")

3. (for DataHub OSS instances) Check GMS (backend) logs for processing errors:

# For DataHub open source
# Check for MCP processing errors
kubectl logs -f deployment/datahub-gms | grep -i "error.*mcp"

# Look for schema validation failures
kubectl logs deployment/datahub-gms | grep "validation"

4. For REST API, use synchronous emission:

# REST API automatically validates and returns errors
emitter = DatahubRestEmitter(
    "http://localhost:8080",
    token="your-token"
)

# Will raise exception immediately if invalid
emitter.emit_mcp(...)

5. Monitor MCP processing metrics:

# Check DataHub metrics endpoint
curl http://localhost:8080/api/v2/system/metrics | jq '.mcp_processing'

MCPs are the recommended way to emit metadata (replaces legacy MCEs). Always use the REST emitter for immediate validation feedback. For high-volume emission, consider using Kafka emitter with proper error handling.

• Metadata Change Proposal
• Python SDK Guide