Skip to main content

Event publishing

The event publisher sends catalog change events to another system as they happen, so a downstream catalog, a data quality service or an audit pipeline can follow tables being created, altered and dropped without polling Gravitino. It delivers to a Kafka topic or to an HTTP endpoint, and every event is the same CloudEvents document either way.

The publisher is an event listener. Its source, build and tests are in the gravitino-event-publisher repository under the Apache License 2.0. It is not part of the Datastrato Enterprise image, so an install that uses it runs a server image with the publisher's jar added.

Quick start​

The example publishes to a Kafka topic over SASL_SSL with SCRAM credentials. It uses the RELEASE, NAMESPACE and LICENSE_FILE variables from Install with Helm.

  1. Build the publisher. The build needs JDK 17 or later and runs the publisher's HTTP smoke test.

    Build the publisher jar
    git clone https://github.com/markhoerth/gravitino-event-publisher.git
    cd gravitino-event-publisher
    ./gradlew build
    ls build/libs/gravitino-event-publisher-0.1.0.jar

  2. Build a server image that carries it. The server loads every jar in /opt/gravitino/libs. Start from the Datastrato Enterprise image of the version you run, and push the result to a registry your cluster pulls from.

    Dockerfile
    FROM {registry}/datastrato/enterprise:{version}
    COPY build/libs/gravitino-event-publisher-0.1.0.jar /opt/gravitino/libs/
    Build and push the image
    docker build -t {your_registry}/datastrato/enterprise-events:{version} .
    docker push {your_registry}/datastrato/enterprise-events:{version}

  3. Store the Kafka credentials in a Secret. The publisher reads them from a mounted file, so no password appears in the server's configuration.

    credentials.properties
    jaas=org.apache.kafka.common.security.scram.ScramLoginModule required username="{kafka_user}" password="{kafka_password}";
    truststore_password={truststore_password}
    Create the credentials Secret
    kubectl -n "$NAMESPACE" create secret generic gravitino-kafka-credentials \
      --from-file=credentials.properties \
      --from-file=truststore.p12={kafka_truststore_file}

  4. Configure the release. Add the publisher to the listener list, point it at the topic, mount the Secret, and switch the server to the new image.

    my-values.yaml
    global:
      imageSets:
        gravitino:
          image:
            registry: {your_registry}
            repository: datastrato/enterprise-events
            tag: "{version}"

    gravitino:
      eventListener:
        names: "search,dashboard-metrics,eventPublisher"
      additionalConfigItems:
        gravitino.eventListener.eventPublisher.class: "com.datastrato.gravitino.eventpublisher.EventPublisherListener"
        gravitino.eventListener.eventPublisher.source: "https://gravitino.example.com/prod"
        gravitino.eventListener.eventPublisher.transport: "kafka"
        gravitino.eventListener.eventPublisher.bootstrapServers: "{broker_1}:9093,{broker_2}:9093"
        gravitino.eventListener.eventPublisher.topic: "gravitino-catalog-events"
        gravitino.eventListener.eventPublisher.kafka.security.protocol: "SASL_SSL"
        gravitino.eventListener.eventPublisher.kafka.sasl.mechanism: "SCRAM-SHA-512"
        gravitino.eventListener.eventPublisher.kafka.config.providers: "file"
        gravitino.eventListener.eventPublisher.kafka.config.providers.file.class: "org.apache.kafka.common.config.provider.FileConfigProvider"
        gravitino.eventListener.eventPublisher.kafka.sasl.jaas.config: "${file:/etc/gravitino/kafka/credentials.properties:jaas}"
        gravitino.eventListener.eventPublisher.kafka.ssl.truststore.location: "/etc/gravitino/kafka/truststore.p12"
        gravitino.eventListener.eventPublisher.kafka.ssl.truststore.type: "PKCS12"
        gravitino.eventListener.eventPublisher.kafka.ssl.truststore.password: "${file:/etc/gravitino/kafka/credentials.properties:truststore_password}"
      extraVolumes:
        - name: kafka-credentials
          secret:
            secretName: gravitino-kafka-credentials
      extraVolumeMounts:
        - name: kafka-credentials
          mountPath: /etc/gravitino/kafka
          readOnly: true

    Each ${file:...} reference reads one key from credentials.properties in the mounted Secret. Keep search and dashboard-metrics in the listener list; replacing the list instead of extending it removes them.

  5. Apply it and confirm the publisher started. Run the install command from Install with Helm again. The server log reports the transport the publisher chose.

    Find the publisher in the server log
    kubectl -n "$NAMESPACE" exec deploy/"$RELEASE-gravitino" -c gravitino -- \
      sh -c 'grep -E "event publisher (initialised|started)|Kafka transport initialised" /opt/gravitino/logs/$HOSTNAME/gravitino-server.log'

  6. Create a table and read the topic. Each change arrives as one record. client.properties holds your own consumer's connection settings.

    Read the topic
    kafka-console-consumer.sh --bootstrap-server {broker_1}:9093 \
      --consumer.config client.properties \
      --topic gravitino-catalog-events --from-beginning --property print.key=true

What is published​

Events​

By default the publisher publishes changes to tables, schemas, filesets and models: objects created, altered, dropped or purged, and model versions linked. The operations property narrows or widens the list. Operations that failed are left out, so a failed create never tells a consumer that an object exists. Read traffic such as loading or listing a table is not in the default list and should stay out of it, since it would bury the changes.

A created or altered table carries the table's full state after the operation: its columns with their types and comments, its properties, its owner and who last changed it. A consumer can apply the state as given, with no merge, which makes a replayed event harmless. An alter also carries the list of changes for a consumer that wants them. A dropped table carries no state, because the server's drop event has none, so a consumer deletes by the event's subject. Changes to schemas, filesets and models arrive as notices that name the object and the change, without its state; a consumer that needs the state loads the object from the Gravitino REST API.

Format​

Every event is a CloudEvents 1.0 JSON document.

A table created event
{
  "specversion": "1.0",
  "id": "a34cc863-d442-476c-8cc9-576acfc4dc52",
  "source": "https://gravitino.example.com/prod",
  "type": "org.apache.gravitino.table.created",
  "time": "2026-10-18T07:23:40.372Z",
  "datacontenttype": "application/json",
  "subject": "enterprise.analytics.sales.daily_orders",
  "sequence": 1,
  "actor": "dana.okafor@example.com",
  "eventsource": "gravitino_server",
  "metalake": "enterprise",
  "catalog": "analytics",
  "data": {
    "entityType": "table",
    "fqn": "enterprise.analytics.sales.daily_orders",
    "columns": [
      {"name": "customer_id", "type": "string", "nullable": false, "position": 0}
    ],
    "owner": "analytics-eng"
  }
}
AttributeContent
typeThe object and the change, such as org.apache.gravitino.table.created or org.apache.gravitino.schema.dropped.
subjectThe object's full name, such as {metalake}.{catalog}.{schema}.{table}. Key consumer state on this.
sourceThe source property, which identifies the server. Consumers fed by more than one server deduplicate on source and id.
actorThe user who made the change, from the request's credentials.
sequenceA counter that increases with each event the server publishes, starting again at 1 when the server restarts.
dataA table's state after the operation. Absent on a drop and on changes to other objects.

Kafka records​

The record value is the event document. The record key is the event's subject, so every event for one object lands on the same partition in the order the changes were made, and a consumer applying the state never sees an older version after a newer one. Each record carries a content-type header of application/cloudevents+json; charset=UTF-8.

HTTP requests​

Each event is one POST with the document as its body, a Content-Type of application/cloudevents+json; charset=utf-8, and an Idempotency-Key header holding the event id, which stays the same across retries of one event.

Configuration​

Every property is set under gravitino.additionalConfigItems in my-values.yaml, prefixed with gravitino.eventListener.eventPublisher., as a quoted string. The name eventPublisher is the one added to gravitino.eventListener.names; any name works if it matches everywhere.

All transports​

PropertyDescriptionDefault
classcom.datastrato.gravitino.eventpublisher.EventPublisherListener. Required.(none)
transportkafka or http.http
sourceA stable URI for this server, unique per server and unchanged across restarts.(none)
operationsComma-separated operation types to publish, in lower case, such as create_table,alter_table,drop_table.Table, schema, fileset and model changes
includeFailuresAlso publish operations that failed.false

Kafka​

PropertyDescriptionDefault
bootstrapServersComma-separated brokers. Required.(none)
topicThe topic to write to. Required. The publisher does not create it.(none)
kafka.*Passed to the Kafka producer with the kafka. prefix removed. TLS, SASL and config providers are set here.(none)
deliveryTimeoutMsHow long the producer keeps retrying one event before it is given up.30000
requestTimeoutMsHow long the producer waits for one broker response.10000
maxBlockMsHow long sending waits for broker metadata when no broker answers.5000

The connection must authenticate. The publisher refuses to start unless kafka.security.protocol is SASL_SSL with kafka.sasl.jaas.config set, or SSL with a client keystore in kafka.ssl.keystore.location. The producer writes with acks=all, retries without duplicating a record, and compresses with gzip.

HTTP​

PropertyDescriptionDefault
urlThe https endpoint to POST to. Required.(none)
auth.typeoauth2, mtls or bearer. Required.(none)
auth.oauth2.tokenEndpoint, auth.oauth2.clientId, auth.oauth2.clientSecret, auth.oauth2.scopeOAuth client credentials the publisher exchanges for a token.(none)
auth.mtls.keyStorePath, auth.mtls.keyStorePassword, auth.mtls.trustStorePath, auth.mtls.trustStorePasswordClient certificate for mutual TLS.(none)
auth.bearer.tokenA static token, for an endpoint that accepts nothing else.(none)
header.{name}A header sent on every request.(none)
maxAttemptsAttempts per event before it is given up.4
initialBackoffMs, maxBackoffMsWait before the first retry, doubling up to the maximum.250, 5000
connectTimeoutMs, requestTimeoutMsConnection and per-request timeouts.5000, 10000

The HTTP transport reads its credentials only from these properties, which the chart writes into the server's ConfigMap in plain text. Restrict who can read ConfigMaps in the release's namespace when you use it. Kafka avoids this through the file references in the quick start.

Delivery​

Delivery is at most once. The publisher runs on a queue of its own, so a slow or unreachable destination never slows a catalog operation, but an event can be lost in three ways:

  • The queue is full when the event arrives. The server drops it and logs a warning; see Queue behavior. Raise gravitino.eventListener.queueCapacity if the warning appears.
  • The server stops with events still queued.
  • Delivery keeps failing until its retries run out. The event is then written to the server log under the logger com.datastrato.gravitino.eventpublisher.DeadLetter, with its full payload. Alert on that logger.

A consumer that must hold a complete copy reconciles against the Gravitino REST API from time to time, and treats the event stream as the fast path rather than the record of truth.

Last updated on