DataHub Actions Filtering Behavior: Client-Side Filtering, lookback_days Semantics, and Event Emission Timing – DataHub Customer Support Portal

```html

Area: Ingestion Issues

Sub-Area: DataHub Actions / Entity Change Events

Issue

Users building incremental ingestion workflows with DataHub Actions and EntityChangeEvent_v1 events encounter three related behavioral surprises: (1) applying multiple filters in the Actions YAML (e.g., combining entityType, operation, and category) causes significant latency or results in no events being returned at all, even when matching events are confirmed to exist; (2) the lookback_days parameter returns events whose auditStamp.time is older than the configured window; and (3) it is unclear whether Entity Change Events are emitted in real-time during ingestion or only after ingestion completes.

You Might Be Asking

Why does adding more filters to my DataHub Actions config cause events to stop appearing?
Why do I receive events with audit timestamps older than my lookback_days setting?
Should I run DataHub Actions before, during, or after my ingestion pipeline?
Is filtering in DataHub Actions applied server-side or client-side?
How can I verify that all pending Kafka writes have been processed in DataHub Cloud?

Solution

1. Filter Complexity and Event Retrieval (Client-Side Filtering)

All filtering in DataHub Actions is applied client-side, not server-side. The backend polling endpoint (/openapi/v1/events/poll) returns all events without any filter parameters. The FilterTransformer in the Actions framework discards events locally after they are received.

In versions prior to the fix introduced in PR #17593, a bug in the matching logic caused events to be incorrectly dropped when filtering on fields such as operation or category:

If an event field value was not a plain string (e.g., None, an integer, or a nested object), the list-match function immediately returned False and the event was dropped.
JSON serialization round-trips (as_json() → json.loads()) could lose or transform fields, causing fields like operation or category to be absent in the deserialized dict.
Variability in EntityChangeEvent structure across different event categories contributed to intermittent behavior.

Short-term workaround (all versions): Use the minimal filter needed. A single event.entityType filter is reliable. Apply additional field-level filtering inside your custom action's Python logic.

name: "datahub-schema-change-action"
source:
  type: "datahub-cloud"
  config:
    lookback_days: 1
filter:
  event_type: "EntityChangeEvent_v1"
  event:
    entityType: "dataset"
action:
  type: "hello_world"
datahub:
  server: "https://<your-instance>.acryl.io"
  token: "<your-datahub-access-token>"

Then filter by operation and category inside your action handler:

def act(self, event: EventEnvelope) -> None:
    change_event = event.event
    if change_event.operation not in ("ADD", "MODIFY"):
        return
    if change_event.category != "TECHNICAL_SCHEMA":
        return
    # Process the event
    ...

Long-term fix: Upgrade to a version of acryl-datahub-actions that includes PR #17593. This release introduces:

A proper Filter abstraction (EventTypeFilter) that runs before transformers.
Direct to_obj() conversion instead of a lossy JSON round-trip, preserving all field types.
Robust nested field matching with correct type handling.
A new list-based filter configuration format with clearer semantics.

With the new filter format:

name: "datahub-schema-change-action"
source:
  type: "datahub-cloud"
  config:
    lookback_days: 1
filters:
  - type: "event_type"
    config:
      filter:
        EntityChangeEvent_v1:
          event:
            - entityType: "dataset"
              operation: ["ADD", "MODIFY"]
              category: "TECHNICAL_SCHEMA"
action:
  type: "hello_world"
datahub:
  server: "https://<your-instance>.acryl.io"
  token: "<your-datahub-access-token>"

2. Understanding `lookback_days` — Kafka Ingestion Time vs. Audit Timestamp

The lookback_days parameter seeks the Kafka consumer offset to a point N days in the past using Kafka's offsetsForTimes API. It uses the Kafka message timestamp (i.e., when the event was written to the Kafka topic), not the auditStamp.time field inside the event payload.

As a result, you may receive events whose auditStamp.time is many days older than the lookback window. This occurs in scenarios such as delayed ingestion, metadata reprocessing, or backfill runs — where a metadata change that originally happened days ago is written to Kafka only recently.

To filter events by their actual audit timestamp, add a time check inside your action handler:

import time

def act(self, event: EventEnvelope) -> None:
    audit_time_ms = event.event.auditStamp.time
    cutoff_ms = int(time.time() * 1000) - (24 * 60 * 60 * 1000)  # 24 hours ago
    if audit_time_ms < cutoff_ms:
        return  # Skip events older than 24 hours by audit timestamp
    # Process the event
    ...

3. Event Emission Timing — Real-Time Streaming During Ingestion

Entity Change Events are emitted in a near real-time, streaming fashion as ingestion results are indexed. The pipeline is:

Ingestion emits MetadataChangeProposal (MCP) events to Kafka.
The MCP consumer writes aspects to DataHub's datastores and emits MetadataChangeLog (MCL) events.
The PlatformEventGeneratorHook (running in the MAE consumer) processes MCL events and generates EntityChangeEvent_v1 entries on the PlatformEvent_v1 Kafka topic in near real-time.
Your DataHub Actions pipeline continuously polls PlatformEvent_v1 and receives events as they arrive.

There is no batch delay waiting for ingestion to complete. There are, however, small asynchronous processing delays between each stage (typically a few seconds). The recommended approach is to run DataHub Actions continuously as a long-running process, in parallel with or independently of your ingestion pipelines.

To verify that all pending Kafka writes have been fully processed (zero lag) in DataHub Cloud, use the Operations API:

curl -X 'GET' \
  'https://<your-instance>.acryl.io/openapi/operations/kafka/mcp/consumer/offsets?skipCache=true&detailed=true' \
  -H 'accept: application/json' \
  -H 'Authorization: Bearer <your-datahub-access-token>'

A response with "totalLag": 0 across all partitions confirms all pending writes have been processed:

{
  "<consumer-group-id>": {
    "<topic-name>": {
      "partitions": {
        "0": { "offset": 12345, "lag": 0 },
        "1": { "offset": 12346, "lag": 0 }
      },
      "metrics": {
        "maxLag": 0,
        "medianLag": 0,
        "totalLag": 0,
        "avgLag": 0
      }
    }
  }
}

Note: This endpoint requires Admin-level access. You can also view lag visually at https://<your-instance>.acryl.io/admin/dashboard/default.

Additional Notes

The client-side filtering bug affecting multi-field filters (combining operation, category, and entityType) was present in acryl-datahub-actions version 1.3.1.5 and earlier. The root cause was a fragile type check in the matches_list() utility that immediately returned False for any non-string field value, combined with potential field loss during JSON serialization round-trips. PR #17593 resolves all of these issues. Until you upgrade, the safest approach is to use a single, minimal filter and perform additional field-level filtering in your action handler code. The lookback_days behavior (Kafka offset time vs. audit timestamp) is by design and requires application-level handling if strict audit-time filtering is needed.