krickert opened a new pull request, #11997: URL: https://github.com/apache/gravitino/pull/11997
### What changes were proposed in this pull request? Gravitino can catalog Kafka topics today, but the catalog only shows that a topic exists, not what is inside it. The messages are Protobuf, Avro, or JSON, usually with a schema registered somewhere, and none of that appears in the catalog. In practice, an engineer who finds a topic in Gravitino still has to dig through a schema registry, a wiki, or tribal knowledge to learn what the records look like. This PR lets a topic carry its message schemas as catalog metadata. For example, an `orders` topic can record that its value is the Protobuf message `com.example.Order`, registered under the `order-value` subject at a given registry URL, and that its key is a JSON-encoded order id. A discovery tool can then show what a topic contains without leaving Gravitino, and a governance process can flag topics that have no declared schema at all. Concretely, this implements the reserved `DataLayout` placeholder so topics can carry named message schema metadata, conventionally `key` and `value`. Each layout can describe a registry reference (`schemaUri`, `schemaSubject`, `schemaVersion`, `schemaId`), inline schema text, a type name, and vendor-specific properties. - **API**: `DataLayout` gains schema fields (all optional, default-method based); new immutable `SchemaDataLayout` implementation (builder-based, `toString` redacts inline schema text); `DataLayouts` helper class (conventional names, defensive copies, ordered change application); `Topic#dataLayouts()`; `TopicChange.updateDataLayout / removeDataLayout / removeDataLayouts`; `TopicCatalog#createTopic` now accepts `Map<String, DataLayout>`. - **REST / client**: `dataLayouts` on topic create request and topic DTO; three new `TopicUpdateRequest` types (`updateDataLayout`, `removeDataLayout`, `removeDataLayouts`); Java client create/alter/read support; OpenAPI spec. - **Validation** (server-side, before any catalog mutation): layouts must be non-empty; schema fields must not be blank; `schemaUri` must be a valid URI; `schemaText` must be valid JSON when format is `avro` or `json`; the reserved internal storage key is rejected from client-supplied topic properties. - **Core / storage**: layouts persist in Gravitino's entity store inside `topic_meta` properties JSON under a reserved key, and are never exposed via `Topic#properties()`. Partial updates preserve untouched layouts. Malformed stored layout JSON logs a warning instead of failing topic loads. Event listeners observe layouts via `TopicInfo`. - **Kafka catalog**: accepts layouts as entity-store-only metadata. Layout-only alters are no-ops against the broker, since Kafka has no broker-side schema storage. Layouts are declarative metadata in this PR. The fields map directly onto schema registry concepts (Confluent Schema Registry, Apicurio, AWS Glue), so a layout can fully reference a registered subject and version, but Gravitino does not connect to a registry here. It does not validate that a referenced subject exists and does not sync when the registry changes. See the follow-up section at the end. ### Why are the changes needed? Topic message schemas are the missing piece of messaging metadata. Without them, Gravitino can answer "what topics exist" but not "what data flows through them", which limits its value for discovery, lineage, and governance over streaming data. Layouts are a named map because messaging payloads carry separate key and value schemas (for example, Schema Registry's `<topic>-key` and `<topic>-value` subjects). A singular layout cannot express the key schema, which is the most common real-world case, and would force a breaking migration of a released contract later. Additional names are allowed for vendor-specific roles. **Intentional API evolution:** the `dataLayout` parameter of `TopicCatalog#createTopic` becomes `Map<String, DataLayout>`. On `main`, `DataLayout` is an `@Unstable` marker interface with no methods, documented as "currently not implemented, only reserved as a placeholder", and the parameter was documented as "always null". `@Unstable` disclaims compatibility across releases, and `TopicCatalog` itself is `@Evolving`. Callers passing the documented bare `null` still compile against the new signature. A deprecated single-layout compatibility overload was considered and deliberately deferred: with both overloads present, the historical bare-`null` calling pattern becomes ambiguous at compile time. It can be added as a targeted follow-up if binary descriptor compatibility is prioritized. Fix: #11996 ### Does this PR introduce _any_ user-facing change? Yes. 1. `TopicCatalog#createTopic(NameIdentifier, String, Map<String, DataLayout>, Map<String, String>)`: third parameter reshaped from the unimplemented `DataLayout` placeholder (see evolution note above). External `TopicCatalog` implementations must update the signature. 2. New public API: `Topic#dataLayouts()`, `SchemaDataLayout`, `DataLayouts`, `DataLayout` schema accessors, and three new `TopicChange` types. 3. REST: optional `dataLayouts` field on topic create/response; new `@type` values `updateDataLayout`, `removeDataLayout`, `removeDataLayouts` for topic updates. All additive. The `format` enum uses lowercase on the wire and is matched case-insensitively on requests. 4. New reserved topic property key `gravitino.internal.data-layouts` (internal storage only; rejected if supplied by clients). 5. No configuration keys added or removed. No storage schema (DDL) change: layouts serialize into the existing `topic_meta` properties JSON. ### How was this patch tested? New unit tests across every affected layer (about 1,200 test LOC): - `api`: `TestDataLayouts`, `TestSchemaDataLayout`, `TestTopicChange`: helpers, defensive copies, ordered change application, argument validation, equals/hashCode, `toString` redaction. - `common`: `TestDataLayoutDTO`: JSON round-trips, case-insensitive format parsing, empty/blank/URI/schema-text validation, nested request validation, reserved-property rejection. - `core`: `TestTopicOperationDispatcher`: create/alter with layouts, partial-update preservation, ordered changes, validation before catalog mutation. `TestPOConverters`: persistence round-trip, reserved-key stripping, malformed-JSON tolerance. - `server`: `TestTopicOperations`: REST create/update/remove happy paths plus 400 responses for invalid layouts and reserved properties (verified the dispatcher is never reached). - `catalog-kafka`: `TestKafkaCatalogOperations`: layouts on create, layout-only alter as broker no-op. - `client-java`: `TestMessagingCatalog`: client create/alter/read with layouts. Verified with `./gradlew :api:test :common:test :core:test :server:test :clients:client-java:test :catalogs:catalog-kafka:test -PskipITs`, plus `spotlessJavaCheck`, javadoc on all touched modules, and `./gradlew :docs:build` for OpenAPI spec validation. ### Follow-up work The intent is to integrate this with schema registries, with Confluent Schema Registry first since it is the de facto standard. The natural next step is a `schema.registry.url` property on the Kafka catalog that lets Gravitino enrich topics automatically: importing key and value layouts from registry subjects when topics are loaded or imported, optionally validating that a referenced subject and version exist when a user writes a layout by hand, and eventually detecting drift between the catalog and the registry. This PR intentionally lands only the data model, which is the storage format that integration will populate, so the registry connectivity design (auth, caching, registry-down behavior) can be reviewed on its own. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
