Structured Properties Delayed or Missing Due to Async Write Queue Congestion – DataHub Customer Support Portal

```html

Area: Ingestion Issues

Sub-Area: Structured Properties / Async Write Queue

Issue

During large-scale ingestion runs, structured property writes can experience significant delays — sometimes several hours — before appearing in the DataHub UI or being returned by the API. This happens because structured property writes share an asynchronous proposal queue (the MCE consumer pipeline) with high-volume ingestion jobs such as BigQuery or other data platform connectors. When a large ingestion run is in progress, structured property write proposals are queued behind a large backlog of ingestion events. Once the queue eventually catches up, the properties do persist, but the delay can be severe enough to appear as data loss to end users. In more extreme cases, Kafka consumer group evictions caused by slow processing can also result in MCL (Metadata Change Log) emission failures, causing the search index to fall out of sync with the primary database even after writes succeed.

Error Messages

Failed to produce MCLs
Offset commit failed — consumer group eviction
404 Not Found — /aspects/{entityUrn}?aspect=structuredProperties&version=0 (transient; may resolve once indexing catches up)

You Might Be Asking

Why do structured properties take hours to appear in the UI even though ingestion seems successful?
Are structured property writes being lost, or are they just delayed?
Why does the /aspects API return a 404 or stale data for an entity that was recently updated with structured properties?
How can I make structured property writes appear immediately without waiting for the ingestion queue to drain?
What is the difference between the database write and the search index update for structured properties?

Solution

Understand the write path.
Structured property writes follow a two-step path in DataHub:
1. The aspect is written to the primary database (e.g., PostgreSQL or MySQL).
2. A Metadata Change Log (MCL) event is emitted to Kafka, which triggers the search index (Elasticsearch) update.
When writes are submitted asynchronously (async=true, the default), the proposal is placed on the MCE (Metadata Change Event) Kafka topic and processed by the MCE consumer. If that consumer is already busy processing a large ingestion run, structured property proposals sit in the queue until the backlog clears.
Confirm whether data exists in the database vs. the search index.
If the UI or a search-backed API call does not show a structured property, first verify whether the write actually reached the primary database by calling the aspects endpoint directly:
```
GET /gms/aspects/{encodedEntityUrn}?aspect=structuredProperties&version=0
```
Replace {encodedEntityUrn} with the URL-encoded URN of the entity, for example:
```
GET /gms/aspects/urn%3Ali%3Adataset%3A%28urn%3Ali%3AdataPlatform%3Abigquery%2C%3Cyour-dataset%3E%2CPROD%29?aspect=structuredProperties&version=0
```
- If this returns data, the write succeeded and the delay is a search index lag. The UI will update once the MCL is processed.
- If this returns a 404, the write itself did not reach the database. The structured property MCP (Metadata Change Proposal) should be re-emitted.
Switch structured property writes to synchronous mode.
The fastest and most reliable fix is to set async=false on structured property write calls. Synchronous writes bypass the MCE queue entirely: the proposal is processed inline, the aspect is written to the database immediately, and the MCL is emitted before the call returns. This eliminates queue-induced delays entirely for structured properties.

Using the DataHub Python SDK:
```
from datahub.emitter.rest_emitter import DatahubRestEmitter

emitter = DatahubRestEmitter(gms_server="https://<your-instance>.datahubproject.io/gms")

# Emit the structured property MCP synchronously
emitter.emit(mcp, async_flag=False)
```
Using the REST API directly (query parameter):
```
POST /gms/aspects?action=ingestProposal&async=false
Content-Type: application/json

{
  "proposal": {
    "entityType": "dataset",
    "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:bigquery,<your-project>.<your-dataset>.<your-table>,PROD)",
    "aspectName": "structuredProperties",
    "changeType": "UPSERT",
    "aspect": {
      "value": "{ ... }",
      "contentType": "application/json"
    }
  }
}
```
High-volume ingestion pipelines (e.g., BigQuery, Snowflake connectors) should remain on the default async=true mode, as synchronous writes at very high volume are slower per entity. Only structured property writes — which are typically lower volume — need to be switched to synchronous mode.

Re-emit structured property MCPs for entities that appear stale.

If specific entities have structured properties that are confirmed present in the database (step 2 returns data) but are not reflected in the UI, those entities need their search index refreshed. Re-emitting the structured property MCP for the affected entities will trigger a new MCL and update the search index.

# Example: re-emit a structured property MCP for a specific entity
from datahub.metadata.schema_classes import (
    MetadataChangeProposalWrapper,
    StructuredPropertiesClass,
    StructuredPropertyValueAssignmentClass,
)
from datahub.emitter.rest_emitter import DatahubRestEmitter

emitter = DatahubRestEmitter(gms_server="https://<your-instance>.datahubproject.io/gms")

entity_urn = "urn:li:dataset:(urn:li:dataPlatform:bigquery,<your-project>.<dataset>.<table>,PROD)"

structured_props = StructuredPropertiesClass(
    properties=[
        StructuredPropertyValueAssignmentClass(
            propertyUrn="urn:li:structuredProperty:<your-property-name>",
            values=["<your-value>"]
        )
    ]
)

mcp = MetadataChangeProposalWrapper(
    entityUrn=entity_urn,
    aspect=structured_props,
)

# Use async=False to bypass the queue
emitter.emit(mcp, async_flag=False)

For DataHub Cloud deployments: contact support for MCE consumer scaling.
If switching to synchronous writes is not immediately feasible and the queue congestion is severe, contact DataHub support to request an increase in MCE consumer replicas for your namespace. This increases the throughput of the async queue and reduces processing lag for all write types. Note that MCE consumer scaling may depend on the health and version of your Elasticsearch cluster, so this option may have prerequisites in your environment.

Additional Notes

Data loss vs. indexing lag: In most cases of structured property delays caused by queue congestion, no data is lost. The primary database write succeeds before the async proposal is enqueued; only the search index update is delayed. Verify using the /gms/aspects endpoint (step 2 above) before concluding that data has been lost.

MCL emission failures: Under sustained queue pressure, the Kafka MCE consumer can be evicted from its consumer group due to processing timeouts. This can cause MCL emission failures where the database write succeeds but the search index is never updated. These failures are logged as Failed to produce MCLs in GMS logs. Affected entities require a re-index or re-emit to recover. A retry topic (FMCE) exists, but it may also become congested under heavy load.

API reads and search index dependency: Most DataHub API endpoints (including /entitiesV2 and the GraphQL API) read from the Elasticsearch search index rather than the primary database. This means that even a successful database write will not be reflected in API responses until the MCL is processed and the index is updated. The /gms/aspects endpoint reads directly from the database and is not affected by search index lag.

Performance trade-off: Synchronous writes (async=false) are slightly slower per individual call than asynchronous writes because they block until the full write path completes. For low-to-medium volumes of structured property writes, this trade-off is acceptable and the latency is negligible. Do not switch high-volume ingestion connectors to synchronous mode.

Elasticsearch cluster health: A degraded or under-resourced Elasticsearch cluster amplifies both MCL emission failures and consumer group eviction rates. Keeping Elasticsearch up to date and properly sized is important for maintaining a healthy async pipeline.

Issue

Error Messages

You Might Be Asking

Solution

Additional Notes

Related Documentation