fix(actions): handle Avro JSON encoding in cloud event source and harden ack manager

[shirshanka]

Summary

• Add _unwrap_avro_json and _fix_avro_bytes helpers to handle Avro JSON union encoding in the cloud event source. Events may arrive Avro-encoded ({"typeName": value} wrappers) which the Pegasus from_json parser rejects; these helpers normalize them to plain dicts before parsing.
• Use .get("name") with fallback in handle_pe instead of direct key access to prevent KeyError when the event name is at the outer envelope level.
• Reduce poll timeout from 2s to 1s and add a 0.1s idle sleep for improved responsiveness.
• Harden the ack manager: use dict.pop(key, None) to avoid KeyError on duplicate acks, and downgrade the "not processed" log from warning to debug (filtered/buffered events are normal).

Test plan

• Verify cloud event source correctly parses both Avro-encoded and plain JSON events
• Verify ack manager handles duplicate ack calls without raising

🤖 Generated with Claude Code

---

Summary by cubic

Handles Avro JSON-encoded cloud events in the event source by parsing them with avro-gen's schema-aware from_obj default mode instead of the Pegasus from_json parser.

• build_metadata_change_log_event and handle_pe now use MetadataChangeLogClass.from_obj and PlatformEventClass.from_obj, which resolve Avro union wrappers ({"typeName": value}) natively.
• Adds unit tests covering Avro-wrapped MCL and platform event payloads.

^{Written for commit fb4341f. Summary will update on new commits.}

[cursor]

PR Summary

Overview
Parses cloud events through Pegasus schema classes so Avro JSON union wrappers ({"typeName": value}) are handled by from_obj instead of manual dict shaping.

Metadata change logs are built by json.loads → MetadataChangeLogClass.from_obj → MetadataChangeLogEvent.from_class, replacing MetadataChangeLogEvent.from_json on the raw message.

Platform events use PlatformEventClass.from_obj for the envelope and read name / payload from the typed object; post_json_transform and direct value["name"] access are removed.

Unit tests cover Avro-wrapped MCL payloads and a platform entityChangeEvent end-to-end through handle_pe.

^{Reviewed by Cursor Bugbot for commit fb4341f. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Avro bytes decoded as UTF-8

High Severity

_fix_avro_bytes encodes Avro JSON bytes fields with UTF-8. The Cloud Events API writes records through Avro jsonEncoder, which maps each raw byte to ISO-8859-1. Non-ASCII aspect and platform-event payloads are therefore stored as the wrong bytes, so downstream actions see corrupted descriptions, docs, and other Unicode metadata.

 

^{Reviewed by Cursor Bugbot for commit 7a010c0. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 175baff. Configure here.}

[cursor]

Plain JSON Unicode values get corrupted

Medium Severity

_fix_avro_bytes always latin-1-encodes any string value sitting next to contentType, including already-decoded plain JSON. Non-ASCII aspect or payload text then raises UnicodeEncodeError (characters above U+00FF) or becomes invalid UTF-8 (characters in U+0080–U+00FF), so handle_pe and build_metadata_change_log_event fail and the source exits.

Additional Locations (2)

• datahub-actions/src/datahub_actions/plugin/source/acryl/datahub_cloud_event_source.py#L115-L122
• datahub-actions/src/datahub_actions/plugin/source/acryl/datahub_cloud_event_source.py#L335-L339

 

^{Reviewed by Cursor Bugbot for commit 175baff. Configure here.}

[skrydal]

It effectively makes it slower, I think. Especially in high-volume cases. /poll returns immediately as long as the requested amount of events is prepared to be returned by the server. By default (and in this case) we will reach for 100 events.
So in high-volume cases we will lose additional 0.1 per 100 events. In all cases we are increasing amount of calls sent to the server by a factor of 2.
Why don't we just make poll timeout seconds value configurable and retain it as default 2? You could adjust the value for your case.
What's the reason for sleeping 100ms in here?

[skrydal]

Exception here is non-fatal and (batch_id, msg_id) should be strictly unique. If this pop hits void, we should be notified, and we will be via the exception path. Please revert the default argument here.

[skrydal]

This is more serious then it sounds now. If an action returned processed=False, then it will stay in self.acks (we won't "pop"). Main loop in _poll_and_process_events will actually crash after event_processing_time_max_duration_seconds (it is bad design, I know). At least now we will see the warning. It never happens because we always mark processed as True. Unless you want to fix it, let's not reduce severity of the log here.

[skrydal]

After analyzing this situation I understand that the source of all problems is that the GMS Avro serializer (Java, ExternalEventsService.convertGenericRecordToJson) encodes bytes fields as latin-1 strings (as per the Avro spec!), while our own avro-gen3 (Python) decodes them as utf-8 (against the spec — _primitive_from_json / _primitive_to_json in avrojson.py). Every non-ASCII aspect value crossing that boundary gets corrupted. That mismatch — not the union wrappers — is the actual production bug.

Therefore I propose fixing the serde bug in the serde layer, not here. This PR's _fix_avro_bytes re-implements bytes decoding by recursively pattern-matching dict keys inside an actions plugin — serde logic leaking into yet another layer, and schema-blind on top (it decides "wrapper or data" from key spelling; a single-entry map like {"consumer.id": "x"} gets destroyed, the helper is not idempotent, and its safety silently depends on every map in the MCL/PE envelope schemas staying optional — one non-optional map field added to MetadataChangeLog later turns this into a data-dependent poison-event crash loop no test catches).

Prerequisite: declare the wire contract. The events API has emitted Avro JSON since it was introduced — we never send plain (Pegasus-style) JSON. The plain-format test fixtures describe a format that doesn't exist on the wire, and they are fundamentally incompatible with latin-1 bytes decoding: a plain aspect value containing real Unicode (e.g. "数据集") cannot be latin-1-encoded and breaks deserialization — in this PR and in any latin-1-based approach alike. So step zero is: Avro JSON is the only supported input; delete the plain-format fixtures and regenerate the rest from actual JsonEncoder output.

The fix, in acryldata/avro_gen: add an optional bytes_encoding argument to from_obj / to_obj (threaded from DictWrapper into _primitive_from_json / _primitive_to_json, applied symmetrically), defaulting to utf-8 — today's behavior, zero blast radius. Changing the default is not safe: avro-gen3 is self-consistent (utf-8 in both directions), so every Python↔Python round-trip and every previously serialized artifact depends on the current convention. Spec-compliant decoding becomes an explicit opt-in at the boundary that needs it. Release, bump the pin in OSS, done — both repos are ours.

Then this code becomes pure consumption, with no serde logic at all — from_json in event_registry.py (also ours) forwards the kwarg to from_obj, and the event source goes back to the pre-PR one-liner plus one argument:

def build_metadata_change_log_event(msg: ExternalEvent) -> MetadataChangeLogEvent:
    return MetadataChangeLogEvent.from_json(msg.value, bytes_encoding="iso-8859-1")

No _unwrap_avro_json, no _fix_avro_bytes, no tuples=True: the default avro-gen3 converter already resolves Avro JSON union wrappers schema-aware (it matches the single-key dict against the union's declared branches) — tuples=True is what disabled that and created the need for the heuristic in the first place. And handle_pe becomes symmetric instead of hand-parsing dict keys:

    pe = PlatformEventClass.from_obj(raw, bytes_encoding="iso-8859-1")
    if pe.name == ENTITY_CHANGE_EVENT_NAME: ...

PlatformEvent needs no special handling: it has zero unions (all three fields required), so the encoder never wraps anything in it.

One open question: the test fixtures wrap required fields ("name": {"string": ...}, "entityType": {"string": ...}). The Avro JsonEncoder cannot produce that from these schemas — only union (optional) fields get wrapped. If some component in the gms-lite stack really emits this shape, please share a wire capture; that producer violates its own schema and should be fixed there, not compensated for in every client. If no capture exists, these fixtures should go.

[shirshanka]

No longer needed — the existing from_json on master handles Avro JSON unions correctly. The real fix (missing sourceDetails in documentation ECEs) is in #19936.