fix(ingestion): decode Kafka sample data per topic and exclude internal topics - #30897
fix(ingestion): decode Kafka sample data per topic and exclude internal topics#30897IceS2 wants to merge 5 commits into
Conversation
…al topics Sample data only ever worked for Avro. The consumer was built with AvroDeserializer hard-wired as its value deserializer for every topic, so JSON and plain-text payloads failed the Confluent magic-byte check and were dropped with a warning. The consumer was also created only when a schema registry was configured, which meant generateSampleData silently did nothing without one. The consumer now hands back raw bytes and decode_message picks the decoder from the topic's own schema type, which it already knew how to do. That removes the registry from the consumer's construction path and makes every schema type work. Confluent framing is stripped before a non-Avro payload is read, and the decode is strict so an undecodable binary payload is skipped by the caller rather than stored as mojibake. Avro deserializers are cached per schema in a bounded LRU, since decode_message runs once per sampled message. Verified against a two-broker cluster with a registry: all four decodable topics now return sample data where only the two Avro ones did before. Without a registry, JSON and text topics return data and Avro topics return nothing with a clear per-message warning. Also give topicFilterPattern a default that excludes Kafka's own internal topics. Without it _schemas and __consumer_offsets are ingested as Topic entities and each emits a schema-registry 404 per run. Ingestion never reads the connection-level pattern; the UI seeds it into new agents only, so existing agents keep their stored config and nothing is soft-deleted.
❌ PR checklist incompleteThis PR cannot be merged until the following are addressed on its linked issue:
The fields live on the linked issue in the Shipping project (open the issue → right sidebar → Projects). After you set them, re-run this check (or push a commit) — issue/project changes do not re-trigger it automatically. Maintainers can bypass this check by adding the |
github-actions
Bot
added
Ingestion
safe to test
![gitar-bot[bot]](https://avatars.githubusercontent.com/in/827041?s=60&v=4){kind=small}
There was a problem hiding this comment.
Pull request overview
This PR fixes Kafka/Redpanda ingestion sample-data decoding by switching to raw-byte consumption and decoding per-topic schema type, while also preventing Kafka internal topics from being default-ingested via a new UI-seeded default topicFilterPattern.
Changes:
- Update Kafka/Redpanda connection creation to always build a raw-byte
Consumer(no hard-wired Avro deserializer) and isolate admin vs consumer configs. - Implement strict per-schema-type decoding in
CommonBrokerSource, including Confluent framing stripping for non-Avro payloads and a bounded Avro deserializer cache. - Add default
topicFilterPatternexclusions for Kafka internal topics to Kafka/Redpanda connection schemas, with unit tests for decoding and connection construction.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| openmetadata-spec/src/main/resources/json/schema/entity/services/connections/messaging/redpandaConnection.json | Adds default topic filter exclusions to avoid internal topics being seeded into new configs. |
| openmetadata-spec/src/main/resources/json/schema/entity/services/connections/messaging/kafkaConnection.json | Same default topic filter exclusions for Kafka connections. |
| ingestion/tests/unit/source/messaging/test_common_broker_source.py | Adds unit tests for strict decoding behavior, Confluent framing stripping, and Avro deserializer caching. |
| ingestion/tests/unit/source/messaging/kafka/test_connection.py | Adds unit tests ensuring a consumer is created without Schema Registry and admin config doesn’t inherit consumer-only settings. |
| ingestion/src/metadata/ingestion/source/messaging/kafka/connection.py | Switches from DeserializingConsumer to raw Consumer, decoupling sample collection from Schema Registry and separating admin config. |
| ingestion/src/metadata/ingestion/source/messaging/common_broker_source.py | Adds framing stripping + strict decoding, Avro deserializer LRU cache, and adjusts polling/error handling. |
…r-only payload Review feedback. strip_confluent_framing required the payload to be longer than the header, so a message consisting of framing plus an empty body kept its header and then failed to decode. It now strips at exactly the header length. The poll loop warned on every message carrying an error, including _PARTITION_EOF. That marker never appears under the default config, but consumerConfig accepts arbitrary librdkafka keys, so a user who sets enable.partition.eof would get one spurious warning per partition per topic. EOF now ends that partition quietly and polling continues, since other partitions may still have data. Verified against a two-broker cluster with enable.partition.eof on: no warnings, and sample data still collected on every topic. Also corrected a test docstring that described the Avro path while asserting on the non-Avro one.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.
Suppressed comments (2)
ingestion/src/metadata/ingestion/source/messaging/common_broker_source.py:328
decode_messagestrips the Confluent framing for all non-Avro records (includingSchemaType.Other). That can violate the intended “strict decode” behavior: a binary payload that would fail UTF-8 decoding (and be skipped) may become decodable after the first 5 bytes are removed, causing corrupted sample data to be stored. Stripping should be limited to schema types that actually use Confluent framing (e.g. JSON Schema).
# Strict: a binary payload we cannot type (Avro with no registry configured)
# must be skipped by the caller, not stored as mojibake.
return strip_confluent_framing(bytes(record)).decode("utf-8")
ingestion/src/metadata/ingestion/source/messaging/kafka/connection.py:104
get_connectionnow always constructs aconfluent_kafka.Consumer(even when sample data is disabled). However, the onlyconsumer_client.close()in the codebase is guarded bygenerate_sample_datainCommonBrokerSource.close(), so in non-sample-data runs this consumer will never be closed. This expands an existing lifecycle gap and can leak sockets/background threads over long-running ingestion processes. Consider registeringconsumer_client.close()in the connection lifecycle (e.g., viaBaseConnection._on_close(...)inKafkaConnection._get_client) or lazily creating the consumer only when sample data is enabled.
consumer_config["bootstrap.servers"] = connection.bootstrapServers
consumer_config.setdefault("group.id", "openmetadata-consumer")
consumer_config.setdefault("auto.offset.reset", "largest")
consumer_config["enable.auto.commit"] = False
consumer_client = Consumer(consumer_config)
✅ Playwright Results — workflow succeededValidated commit ✅ 667 passed · ❌ 0 failed · 🟡 0 flaky · ⏭️ 3 skipped · 🧰 0 lifecycle flaky PerformanceBlocking targets: ✅ met · Optimization targets: 🟡 in progress Shard-job maxima below are not the full workflow wall time; the linked run includes build, fixture, planning, and reporting. 🕒 Full workflow signal wall (to summary) 44m 25s ⏱️ Max setup 3m 32s · max shard execution 18m 28s · max shard-job elapsed before upload 22m 29s · reporting 6s 🌐 214.34 requests/attempt · 2.67 app boots/UI scenario · 10.74% common-shard skew Optimization targets still in progress:
How to debug locally# Download playwright-test-results-<shard> artifact and unzip
npx playwright show-trace path/to/trace.zip # view trace |
…ients Constructing a KafkaSource runs get_connection for real, so this test created a librdkafka AdminClient that background-dialled localhost:9092. Building the consumer unconditionally, as this branch now does, made it two clients rather than one. Patching both constructors keeps the test on the SSL wiring it actually asserts. This was the only place in tests/unit that built a real messaging client.
…:open-metadata/OpenMetadata into fix/kafka-sample-data-and-default-filter
Code Review ✅ Approved 1 resolved / 1 findingsDecouples sample data decoding from the schema registry to support JSON and plain-text payloads, caches Avro deserializers, and excludes internal Kafka topics. Addresses the strip_confluent_framing finding. ✅ 1 resolved✅ Edge Case: strip_confluent_framing may corrupt unframed payloads starting with NUL
OptionsDisplay: compact → Showing less information. Comment with these commands to change the behavior for this request:
Was this helpful? React with 👍 / 👎 | Gitar | Powered by Gitar — free for open source |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 7 out of 7 changed files in this pull request and generated no new comments.
Suppressed comments (1)
ingestion/src/metadata/ingestion/source/messaging/kafka/connection.py:104
get_connectionnow always instantiates aconfluent_kafka.Consumer. HoweverMessagingServiceSource.close()only callsself._connection.close()(BaseConnection), andKafkaConnectiondoes not register any_on_closeteardowns for the underlyingKafkaClient. This means the Consumer can remain unclosed (background threads/sockets) for the lifetime of the process, and the impact is now unconditional even when sample data is disabled.
Consider registering a teardown in KafkaConnection._get_client() (e.g., self._on_close(client.consumer_client.close)) or otherwise ensuring the Consumer is created only when needed and always closed via BaseConnection’s lifecycle.
consumer_config["bootstrap.servers"] = connection.bootstrapServers
consumer_config.setdefault("group.id", "openmetadata-consumer")
consumer_config.setdefault("auto.offset.reset", "largest")
consumer_config["enable.auto.commit"] = False
consumer_client = Consumer(consumer_config)
Three related defects found while exercising the Kafka connector end to end.
Sample data only ever worked for Avro
get_connectionbuilt aDeserializingConsumerwithAvroDeserializerhard-wired asvalue.deserializerfor every topic, regardless of its schema type. JSON and plain-text payloads failed the Confluent magic-byte check and were dropped:The consumer now hands back raw bytes and
decode_messageselects the decoder from the topic's ownmessageSchema.schemaType— which it already had branches for. Confluent framing is stripped before a non-Avro payload is read, so a topic written with a JSON Schema serializer is readable too.The decode is strict. An undecodable binary payload raises, and the caller already logs a warning and skips that message. That matters: with
errors="replace"an Avro payload that cannot be deserialised is stored as mojibake, which is worse than storing nothing.generateSampleDatasilently did nothing without a Schema RegistryThe consumer was constructed only inside
if connection.schemaRegistryURL:, so with no registryconsumer_clientwasNone, the poll was skipped, and an emptyTopicSampleDatawas yielded — no warning, toggle apparently honoured. Decoupling the decoder from the consumer removes the registry from that path entirely.Two smaller things fell out:
admin_client_configwas the same dict object asconsumer_config, so consumer-only settings leaked into it; it is now a copy. And with a plainConsumer,poll()returns messages carrying.error()instead of raising, so that is now checked rather than treated as data.AvroDeserializerwas being constructed once per message; it is now cached per schema in a bounded LRU.Kafka's internal topics were ingested as entities
get_topic_listreturns everything the broker knows andtopicFilterPatternhad no default, so_schemasand__consumer_offsetsbecame Topic entities, each emitting a schema-registry 404 per run. The connection schema already had the field titled "Default Topic Filter Pattern" — it just had nodefault, unlike ten database connectors that shipexcludes: ["^information_schema$"].Added to
kafkaConnection.jsonandredpandaConnection.json:Deliberately narrow — no blanket
^_.*, since a single leading underscore is legal in user topic names, and notconnect-*, whose names are customer-configurable.No migration, and no risk to existing services. Ingestion never reads the connection-level pattern; it reads
sourceConfig.topicFilterPattern. The only consumer of the connection-level value is the UI, which seeds it into a new agent's form (IngestionConfigUtils.ts,isEditMode === false) and ignores it when editing. Existing agents keep their stored config, so nothing starts getting soft-deleted.Verification
Live against a two-broker cluster with a Schema Registry, comparing before and after on the same seeded topics:
om_orders_avroanalytics_clicks_avroom_customers_jsonom_logs_noschemaWith the registry removed from the config, the JSON and plain-text topics still return sample data (previously nothing at all was collected), and the Avro topics return nothing plus a clear per-message warning rather than unreadable text.
Unit tests cover per-schema-type decoding, Confluent framing, the deserializer cache, strict decoding of binary payloads, and that a consumer is built with no registry configured while the admin client does not inherit consumer-only settings.
tests/unit/source/messaging,tests/unit/samplerandtests/unit/topologyare green.Redpanda shares
get_connectionandCommonBrokerSource, so it is fixed by the same change.