Configure a Notification Sink

This guide wires up a notification sink so an external system finds out, in real time, that Invaract checked a write (and what it decided) or that a write actually completed. This is entirely opt-in: with no sink configured, nothing here runs, and the enforcement rule behaves exactly as it would without this guide.

Two events, two moments

ContractValidationEvent fires when ContractEnforcementRule checks a write, before Spark executes anything — a FAILED event means the write never ran. WriteEvent fires later, only once Spark reports a write actually completed, and describes the write itself (location, format, schema, metrics) — a .checkpoint() earlier in the job doesn’t affect it. See Verification vs. Enforcement for why these are two separate mechanisms.

Write a sink configuration file

Sink configuration lives in a plain .properties file, separate from your contract — it’s a deployment concern (an endpoint, a file path, possibly credentials), not something that belongs alongside a contract describing data shape.

notify.properties

sink.enabled=true
sink.class=com.invaract.verification.notification.FileNotificationSink
sink.property.path=/var/log/invaract/events.jsonl

Key	Required	Meaning
sink.enabled	No (default false)	Master switch: true or false. Every other key is ignored when this is false. Any other value is refused, so a typo never silently turns the sink off.
sink.class	Yes, when enabled	Fully-qualified class name of a NotificationSink implementation with a public no-arg constructor.
sink.property.<name>	Depends on the sink	Passed to that sink’s configure as <name> -> value, prefix stripped. FileNotificationSink requires sink.property.path.

Attach it via spark-submit --conf

If your job is installed with InvaractSparkSessionExtension (the zero-code path), a sink attaches the same way:

spark-submit \
  --class com.example.MyRealJob \
  --master yarn \
  --jars invaract-spark-adapter-0.11.0.jar \
  --conf spark.sql.extensions=com.invaract.sparkadapter.InvaractSparkSessionExtension \
  --conf spark.invaract.contract=/path/to/contract.yaml \
  --conf spark.invaract.notifyConfig=/path/to/notify.properties \
  my-real-job.jar

On Apache Beam, the same file attaches with --invaractNotifyConfig=/path/to/notify.properties (Java) or --invaract_notify_config=/path/to/notify.properties (Python). Beam publishes a ContractValidationEvent for every check, before anything runs, but no WriteEvent or JOB_SUMMARY: it has no after-the-fact listener to see a write complete. In dry-run mode it publishes CONTRACT_INFERENCE and DRY_RUN_SUMMARY, as Spark does.

InvaractSparkSessionExtension builds the sink from notify.properties itself, passes it to forContract for ContractValidationEvents, and registers a SparkAdapterListener on the same session for WriteEvents — the two-call wiring Step 2/3 below do by hand, done automatically. Nothing in MyRealJob’s source needs to import com.invaract.verification.notification at all.

Need something spark.invaract.notifyConfig can’t express?

If your job installs Invaract by calling ContractEnforcementRule.forContract directly, wire the sink up the same way — building it from something other than a .properties file, or wrapping it in SummarizingNotificationSink/RetryingNotificationSink before it’s ever registered. See Install and Configure Invaract Explicitly in Code.

What an event looks like

Every event is one JSON object, and its eventType says which. These are real events, copied from real runs; they are published as files next to the JSON Schema. A ContractValidationEvent (CONTRACT_VALIDATION), from a passing check:

{
  "eventType": "CONTRACT_VALIDATION",
  "timestamp": 1792586926573,
  "unverifiableInputs": [],
  "job": {
    "engineDetails": {
      "master": "local[*]",
      "deployMode": "client"
    },
    "name": "InvaractDemoJobHarness",
    "startTimeMs": 1792586919215,
    "attributes": {
      "team": "orders",
      "dag": "orders_daily"
    },
    "jobId": "orders_nightly",
    "engineVersion": "3.5.7",
    "runId": "local-1792586920008",
    "user": "root",
    "engine": "spark"
  },
  "eventId": "d20ce5e99c811308aa7129f026455786eaad88bf832183d700d62e40aea6d6a6",
  "fingerprints": null,
  "status": "PASSED",
  "metadata": {
    "description": {
      "purpose": "Contract for the output of the demo InvaractPlugin pipeline (plugin/src/main/scala/com/invaract/plugin/InvaractPlugin.scala), written by runner/DemoJobHarness.scala to demo/output/result.parquet.\n"
    },
    "owner": "invaract-demo"
  },
  "violations": [],
  "schemaVersion": 1,
  "roleConformance": [],
  "runId": "local-1792586920008",
  "dataQuality": [],
  "contract": "invaract_demo_output@1.0.0"
}

A failed check carries its violations (this one is from the regression pack, which deliberately breaks a contract):

{
  "eventType": "CONTRACT_VALIDATION",
  "timestamp": 1792586962933,
  "unverifiableInputs": [],
  "job": {
    "engineDetails": {
      "master": "local[*]",
      "deployMode": "client"
    },
    "name": "InvaractDemoJobHarness",
    "startTimeMs": 1792586954690,
    "attributes": {
      "team": "orders",
      "dag": "orders_daily"
    },
    "jobId": "orders_nightly",
    "engineVersion": "3.5.7",
    "runId": "local-1792586955529",
    "user": "root",
    "engine": "spark"
  },
  "eventId": "e50bd679a876382a95a6de7498e88bb8edc42af020816ee69b2503e801fbf524",
  "fingerprints": null,
  "status": "FAILED",
  "metadata": {
    "description": {
      "purpose": "A deliberately-broken variant of demo/contracts/invaract_output.yaml, kept for reference/demonstration of ContractEnforcementRule's abort behavior (see ROADMAP.md Phase 5 / docs/SPARK_ADAPTER.md). Not wired into ./dev/test — the real demo pipeline is expected to pass its real contract, not this one. Declares a 'customer_name' output field the real InvaractPlugin pipeline never produces, so verification fails deterministically.\n"
    },
    "owner": "invaract-demo"
  },
  "violations": [
    {
      "location": "file:/home/user/Invaract/demo/output/result_broken_example.parquet",
      "expected": "string",
      "remediation": "Add a 'customer_name' column (type 'string') to the output, or mark it optional in the contract if it isn't always produced.",
      "message": "required field 'customer_name' is absent from the actual OUTPUT schema",
      "column": "customer_name",
      "type": "MISSING_OUTPUT_FIELD"
    }
  ],
  "schemaVersion": 1,
  "roleConformance": [],
  "runId": "local-1792586955529",
  "dataQuality": [],
  "contract": "invaract_demo_output@1.0.0"
}

…and a WriteEvent (WRITE), once the write completes:

{
  "format": "parquet",
  "icebergSnapshotId": null,
  "eventType": "WRITE",
  "catalog": null,
  "location": "file:/home/user/Invaract/demo/output/result.parquet",
  "timestamp": 1792586927566,
  "rowCount": 10,
  "bytesWritten": 1324,
  "job": {
    "engineDetails": {
      "master": "local[*]",
      "deployMode": "client"
    },
    "name": "InvaractDemoJobHarness",
    "startTimeMs": 1792586919215,
    "attributes": {
      "team": "orders",
      "dag": "orders_daily"
    },
    "jobId": "orders_nightly",
    "engineVersion": "3.5.7",
    "runId": "local-1792586920008",
    "user": "root",
    "engine": "spark"
  },
  "eventId": "108aeb8b5b912576ec5f410f02400ab33338eba82c6e2f40638a393d189d9a6e",
  "operation": null,
  "partitionColumns": [],
  "schema": [
    {
      "name": "id",
      "type": "integer",
      "nullable": true
    },
    {
      "name": "value",
      "type": "integer",
      "nullable": true
    },
    {
      "name": "value_squared",
      "type": "integer",
      "nullable": true
    },
    {
      "name": "value_tier",
      "type": "string",
      "nullable": false
    }
  ],
  "fileCount": 1,
  "durationMs": 980,
  "deltaVersion": null,
  "metadata": {
    "description": {
      "purpose": "Contract for the output of the demo InvaractPlugin pipeline (plugin/src/main/scala/com/invaract/plugin/InvaractPlugin.scala), written by runner/DemoJobHarness.scala to demo/output/result.parquet.\n"
    },
    "owner": "invaract-demo"
  },
  "saveMode": "overwrite",
  "schemaVersion": 1,
  "runId": "local-1792586920008",
  "contract": "invaract_demo_output@1.0.0",
  "datasetType": null
}

Dry-run mode, when pointed at a sink, publishes two more event types (CONTRACT_INFERENCE and DRY_RUN_SUMMARY) — see Infer a Starting Contract with Dry-Run Mode.

On the dry-run events, metadata is the draft’s extensions, which come from spark.invaract.contract.metadata.*; once a draft is promoted, the enforcement events below carry the same keys. See Tag the draft, and link events back to the run.

metadata is copied verbatim from your contract’s own extensions: block — see Contract Format — so anything you’ve already recorded there (an owning team, an upstream system) rides along on every event without a second metadata vocabulary to maintain.

runId is the engine’s own identifier for the run (for Spark, the application’s ID, sparkContext.applicationId) on both event types — useful for grouping events from the same job run when a sink aggregates events across many jobs.

job says which engine and which job produced the event, so a sink that receives events from many jobs (or, one day, from more than one engine) can tell them apart and group one job’s events across runs. It is the same object the dry-run events carry: engine ("spark") and engineVersion; jobId, the job’s stable identity across runs, from spark.invaract.jobId (absent if you didn’t set it — set it, because Spark’s own app name is often different on every run under a scheduler); runId, name and user; engineDetails (for Spark, master and deployMode); and attributes, one entry for every spark.invaract.job.metadata.<key>=<value> you set, with the prefix stripped. Nothing else from your Spark configuration is sent. On a JobSummaryEvent it is the job of the events being summarized.

fingerprints on ContractValidationEvent is null unless you’ve opted into semantic lineage fingerprinting — a separate, opt-in feature that computes a deterministic hash of the transformation’s actual business logic, published here alongside the PASS/FAIL verdict so a subscriber can track a transformation’s fingerprint over time, not just whether it currently satisfies its contract’s schema.

dataQuality is [] unless you’ve opted into static data-quality verification with a contract that declares field constraints — populated the same way on both a PASSED and a FAILED check, since most verdicts (Guaranteed/NotGuaranteed/NotStaticallyVerifiable) never block the write on their own; only a Violated entry does, and only that one also shows up in violations.

roleConformance is [] unless you’ve opted into role-consistency checking (spark.invaract.roleConsistency=true) with a contract that declares type: on at least one input — populated the same way on a PASSED and a FAILED check, since most verdicts (Conforms/CannotDetermine) never block the write on their own; only a Contradicts entry does, and only that one also shows up in violations.

unverifiableInputs is [] for the overwhelming majority of checks, and always on — there’s no flag to enable it. A .checkpoint() between a read and the checked write severs Spark’s own analyzed-plan lineage back to that read, so Invaract remembers each plan it sees and, when a checkpointed Dataset appears, looks up the plan it was made from — synchronously, so a write checked right after .checkpoint() is treated no differently from one checked later. A declared input read before a checkpoint is therefore simply read: no violation, and no unverifiableInputs entry. An entry appears only when a declared input has no matching read in the checked plan and the plan still contains a boundary Invaract could not resolve — a checkpoint created before the enforcement rule was installed, one whose origin was evicted from Invaract’s bounded memory of recent plans, one whose origin is ambiguous, a cached relation, or a Delta MERGE whose source is the checkpointed Dataset. An unrelated checkpoint elsewhere in the plan or job never excuses a genuinely missing input (a wrong location, a typo, a forgotten read): that still fails as MISSING_INPUT. Each entry names the input (inputName/inputLocation) and the boundary (unknownNodeTypes, e.g. ["LogicalRDD"]) — report-only, the same as dataQuality/roleConformance above: an entry here never becomes a MISSING_INPUT violation and never blocks the write, since an absence Invaract can’t actually prove must never be treated as a caught defect (the same reasoning RoleConformanceVerdict.CannotDetermine follows — this is deliberately not the fail-closed treatment Fail-Closed by Default describes for a totally unrecognized write, since here only one input’s visibility is uncertain, not the whole write’s meaning).

WriteEvent’s durationMs is always populated, straight from Spark’s own listener callback. rowCount, bytesWritten, and fileCount come from Spark’s internal SQLMetrics and are only populated for plain file-format writes (Parquet, CSV, JSON, ORC) via DataFrameWriter.save/.parquet/etc. They are null for Delta Lake and Iceberg writes — neither connector populates these particular metrics on the plan node Invaract observes, confirmed against a real local write of each. A sink implementation should treat these three fields as always-optional, not assume they’re present just because the write succeeded.

deltaVersion and icebergSnapshotId are the connector’s own identifier for the commit a write just produced — deltaVersion is Some only for a Delta write (null for everything else, Iceberg included), and icebergSnapshotId is Some only for an Iceberg write (null for everything else, Delta included):

{"...": "...", "format": "delta", "deltaVersion": 3, "icebergSnapshotId": null}

{"...": "...", "format": "iceberg", "deltaVersion": null, "icebergSnapshotId": 5964503180593669213}

partitionColumns lists which column(s) the write’s target is partitioned by, when that’s known — [] for an unpartitioned target (as in the example above) or a write shape this isn’t wired up for, never null. For a catalog table partitioned by a derived transform rather than a plain column (Iceberg’s bucket/days/years/months partitioning, for example), each entry is that transform’s own description instead of a bare column name:

{"...": "...", "format": "parquet", "partitionColumns": ["region", "year"]}

{"...": "...", "format": "iceberg", "partitionColumns": ["bucket(4, id)", "days(event_time)"]}

This reports which columns the target is partitioned by, not which specific partition values this particular write’s own rows happened to touch — Invaract doesn’t currently surface that.

datasetType is the contract’s own declared type (DATA_ASSET/SOURCE/CONTROL) for whichever output dataset the write’s location matches (whichever is declared first, if more than one output shares a location) — null whenever there’s no contract at all (dry-run mode), the matching output declares no type:, or no output matches the write’s location:

{"...": "...", "location": "file:/demo/output/result.parquet", "datasetType": "DATA_ASSET"}

Unlike every other WriteEvent field above, this needs no VerificationOptions flag to opt into — it’s a plain lookup against the contract’s own static declarations, not a runtime check, so it’s always populated when a matching output declares a type.

operation distinguishes a MERGE/UPDATE/DELETE from a plain append or overwrite — saveMode is always null for these (in-place mutation isn’t append/overwrite/ignore/ error), so operation is the field that tells them apart:

{"...": "...", "saveMode": null, "operation": "merge"}

{"...": "...", "saveMode": null, "operation": "delete"}

Fields on every event

Every event, of every type, carries three common fields:

Field	Meaning
schemaVersion	The version of this JSON shape: currently 1. A new field, or a new status value, does not change it, so ignore fields you don’t know. Only a removed or renamed field, or a changed type, would.
eventId	The SHA-256 of the event’s own content, as 64 hex characters. It is deterministic: sending the same event again (a retry, or a replay from a dead-letter file) gives the same eventId, so deduplicate on it.
timestamp	Milliseconds since the Unix epoch, when the event was created.

Validate events and build an endpoint to receive them

A JSON Schema covering every event type is published at:

https://mlltx.github.io/Invaract/schemas/notification/v1/event.schema.json

It is one file (JSON Schema 2020-12). Validate any event against it, or point at a single definition, such as event.schema.json#/$defs/ContractInferenceEvent, to generate types in your language. Real example events, one per event type, sit beside it under schemas/notification/v1/examples/. Both are checked on every change: the engine’s own test suite validates every event it sees against the schema, and the docs build validates the schema and the examples with Ajv.

What your endpoint has to do

HttpNotificationSink sends one event per request: a POST to your URL, with Content-Type: application/json, the body being the event, plus any sink.property.header.* headers you configured.

Your endpoint answers	What Invaract does
2xx	The event is delivered.
408, 429, any 5xx, a connection failure or a timeout	Retries with backoff (three attempts by default), then saves it to the dead letter.
Any other status (400, 401, 403, 404, 422, …)	Treats it as a refusal that retrying cannot fix: no retry, saved to the dead letter at once.

Beyond the status codes:

• Answer quickly. The default request timeout is five seconds (sink.property.timeoutMs). Accept the event, then do any slow work afterwards.
• Deduplicate on eventId. Delivery is at-least-once, so the same event can arrive twice.
• Ignore fields you don’t know, and expect new status values. Switch on eventType.
• Don’t assume ordering. Events from many jobs arrive concurrently; order by timestamp.

A complete minimal receiver

This Python receiver (standard library only) refuses what isn’t an Invaract event, accepts what is, and prints each event once:

import json
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer

seen = set()  # eventIds already handled; use a real store in production


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        try:
            event = json.loads(body)
            event_id = event["eventId"]
            event_type = event["eventType"]
        except (ValueError, KeyError, TypeError):
            self.send_response(400)  # a refusal: Invaract will not retry it
            self.end_headers()
            return
        if event_id not in seen:
            seen.add(event_id)
            print(event_type, event.get("status", ""), event_id[:12], flush=True)
        self.send_response(200)  # also for a duplicate: it was already handled
        self.end_headers()

    def log_message(self, *args):
        pass


HTTPServer(("", int(sys.argv[1]) if len(sys.argv) > 1 else 8080), Handler).serve_forever()

Run it, then point the sink at it with sink.property.url=http://localhost:8080/events.

Describing the endpoint in OpenAPI

openapi: 3.1.0
info:
  title: Invaract event receiver
  version: 1.0.0
paths:
  /events:
    post:
      summary: Receive one Invaract notification event
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: https://mlltx.github.io/Invaract/schemas/notification/v1/event.schema.json
      responses:
        '2XX':
          description: Accepted. Answer this for a duplicate as well.
        '429':
          description: Slow down. Invaract retries with backoff.
        '5XX':
          description: Try again later. Invaract retries with backoff.
        '4XX':
          description: Refused. Invaract does not retry; the event is saved to its dead letter.

Built-in sinks

Four ship in spark-adapter itself, all with no extra dependency:

• LoggingNotificationSink — logs each event as one JSON line via SLF4J, at INFO. Needs no sink.property.* configuration.

• FileNotificationSink — appends each event as one JSON line to sink.property.path. This is what the regression pack uses to prove events are published against a real Spark job.

• HttpNotificationSink — POSTs each event’s JSON to sink.property.url, with an optional sink.property.timeoutMs (default 5000). Built on java.net.http.HttpClient, part of the JDK since Java 11 — no extra dependency here either. Requests are sent asynchronously, so a slow or unreachable endpoint doesn’t block your job; a connection failure or non-2xx response is logged, not thrown. Dry-run reporting waits (up to ten seconds) for requests still in flight when the application ends, so a short job doesn’t lose its last events.

  sink.enabled=true
  sink.class=com.invaract.verification.notification.HttpNotificationSink
  sink.property.url=https://example.com/invaract-events

  To authenticate, add request headers with sink.property.header.<Name>, or let the sink send Authorization: Bearer <token> with sink.property.bearerTokenEnv, which names an environment variable to read the token from:

  sink.property.header.X-Tenant=acme
  sink.property.bearerTokenEnv=INVARACT_TOKEN

  Failed requests are retried and, if they still fail, saved to a dead letter: see Delivery.

  The token is read from the environment rather than the properties file on purpose: that file is the kind of thing that gets copied, committed and attached to tickets. A bearerTokenEnv naming a variable that isn’t set — and a malformed header name — fail when the job starts, rather than silently sending unauthenticated requests forever.

• HadoopFsNotificationSink — writes each event as its own JSON object under sink.property.path, on any filesystem Hadoop’s own FileSystem API supports: S3 (s3a://), GCS (gs://, with that connector installed), HDFS (hdfs://), Azure (abfs://), or plain file://. One sink, not one per vendor — the URI scheme picks the backend, exactly like df.write.parquet("s3a://...") already does in your own job. If your contract’s own location: already points at cloud storage, your job’s classpath already carries whatever connector that needs, so this sink adds nothing new to it either.

  sink.enabled=true
  sink.class=com.invaract.sparkadapter.notification.HadoopFsNotificationSink
  sink.property.path=s3a://my-bucket/invaract-events/
  sink.property.hadoop.fs.s3a.access.key=...
  sink.property.hadoop.fs.s3a.secret.key=...

  Every sink.property.hadoop.<key> is set on the underlying Hadoop Configuration — Hadoop’s own config keys, not a separate vocabulary; credentials already configured in your cluster’s core-site.xml apply the normal way and don’t need repeating here.

  One file per event, not an appended log

  S3 objects are immutable — FileSystem.append isn’t reliably supported across backends, and never on S3A. This sink writes a new, uniquely-named object per event instead of appending to one growing file, so it behaves the same way on every backend the path might point at.

Send events to more than one place

FanOutNotificationSink publishes every event to several sinks, configured from the same properties file. sink.property.sinks lists the targets in order; each target x takes x.class and its own x.property.*:

sink.enabled=true
sink.class=com.invaract.verification.notification.FanOutNotificationSink
sink.property.sinks=registry,audit
sink.property.registry.class=com.invaract.verification.notification.HttpNotificationSink
sink.property.registry.property.url=https://registry.corp.internal/events
sink.property.registry.property.bearerTokenEnv=INVARACT_TOKEN
sink.property.audit.class=com.invaract.verification.notification.FileNotificationSink
sink.property.audit.property.path=/var/log/invaract-events.jsonl

An optional x.statuses=A,B limits a target to events whose status is in the list — for dry-run reporting’s inference statuses, or PASSED/FAILED on contract checks. An event with no status (a write, or any summary) always passes. One target failing never stops the others, and a target that can’t be built fails when the job starts, like any single sink does.

Delivery: retries, the dead letter, and duplicates

HttpNotificationSink never blocks your job on the network, and it doesn’t quietly lose an event it couldn’t deliver. These properties (all optional) control that:

Property	Default	Meaning
sink.property.retry.maxAttempts	3	Total tries, counting the first. 1 means no retries.
sink.property.retry.initialBackoffMs	200	Wait before the second try.
sink.property.retry.backoffMultiplier	2.0	Each wait is this many times the last.
sink.property.retry.maxBackoffMs	5000	Upper limit on a single wait.
sink.property.deadLetter.path	none	Where an event goes when it can’t be delivered.
sink.property.deadLetter.class	local file	The sink class that writes the dead letter. Required when the path has a scheme (s3a://, hdfs://, …).

sink.enabled=true
sink.class=com.invaract.verification.notification.HttpNotificationSink
sink.property.url=https://example.com/invaract-events
sink.property.bearerTokenEnv=INVARACT_TOKEN
sink.property.retry.maxAttempts=5
sink.property.deadLetter.path=s3a://my-bucket/invaract-dead-letter/
sink.property.deadLetter.class=com.invaract.sparkadapter.notification.HadoopFsNotificationSink

The dead letter. A plain path is appended to as a local file, one event per line. A path with a scheme (s3a://, hdfs://, gs://, file://) needs a sink that can write there: name sink.property.deadLetter.class=com.invaract.sparkadapter.notification.HadoopFsNotificationSink and it goes through Hadoop’s FileSystem, as one object per event; then sink.property.deadLetter.hadoop.<key> sets Hadoop configuration keys exactly as for HadoopFsNotificationSink. Prefer the scheme form on a cluster: a local file on the driver’s disk disappears with the driver.

When the application ends. Dry-run reporting asks the sink to finish, waiting up to ten seconds, retries included. Whatever is still undelivered at that point is saved to the dead letter right then, because the JVM is about to exit. With no deadLetter.path, it is logged and dropped.

Duplicates. Delivery is at-least-once: an event can be delivered and, a moment later, also saved to the dead letter if the application was ending. That is what eventId is for.

Replaying a local dead-letter file is a loop over its lines; the receiver deduplicates on eventId, so replaying events that did arrive is harmless:

while IFS= read -r event; do
  curl -sS -X POST -H 'Content-Type: application/json' --data "$event" https://example.com/invaract-events
done < /var/log/invaract/dead-letter.jsonl

Summaries and failure-only filtering

Two more classes wrap an existing sink to change what it receives, rather than where events go — useful once you have more than a couple of jobs publishing events and want either a coarser per-job rollup or a page-only-on-failure feed instead of every raw event.

• SummarizingNotificationSink(delegate) forwards every event to delegate unchanged, and also tallies ContractValidationEvent/WriteEvent traffic in memory. Call publishSummary() whenever you want a JobSummaryEvent — totals since construction or the last call — published to delegate, with the counters reset afterward:

  {"checksFailed": 0, "eventType": "JOB_SUMMARY", "timestamp": 1792586928063, "totalViolations": 0, "job": {"engineDetails": {"master": "local[*]", "deployMode": "client"}, "name": "InvaractDemoJobHarness", "startTimeMs": 1792586919215, "attributes": {"team": "orders", "dag": "orders_daily"}, "jobId": "orders_nightly", "engineVersion": "3.5.7", "runId": "local-1792586920008", "user": "root", "engine": "spark"}, "eventId": "da11dc178c97876a4d2484bae3d48ae161cbb75d4753091c0f86c98839ad5ddb", "totalWrites": 1, "durationMs": 8872, "metadata": {"description": {"purpose": "Contract for the output of the demo InvaractPlugin pipeline (plugin/src/main/scala/com/invaract/plugin/InvaractPlugin.scala), written by runner/DemoJobHarness.scala to demo/output/result.parquet.\n"}, "owner": "invaract-demo"}, "schemaVersion": 1, "checksPassed": 1, "runId": "local-1792586920008"}

  A single job (like DemoJobHarness) calls publishSummary() once, right after its one write’s outcome is known — success or failure alike, so “this job failed” is itself part of what the summary reports. A long-running job with many writes can instead call it periodically for a rolling window; nothing about the API changes between the two uses.

• FailureOnlyNotificationSink(delegate) forwards only what’s worth paging someone for: a FAILED ContractValidationEvent, and every JobSummaryEvent (a summary is already a rate-limited, aggregate signal, not raw noise). PASSED checks and every WriteEvent are dropped before reaching delegate.

Both take the sink to wrap as a constructor argument, so build them directly in code around whatever NotificationSinkFactory.create returns — they’re not something sink.class can name on its own, since reflective loading only supports a public no-arg constructor:

import com.invaract.verification.notification.{NotificationSinkFactory, SummarizingNotificationSink}

val configuredSink = NotificationSinkFactory.create(notifyConfig)
val summarizingSink = configuredSink.map(new SummarizingNotificationSink(_))

// ... pass summarizingSink to the enforcement rule and the listener, same as
// any other Option[NotificationSink] ...

// once the job's outcome is known:
summarizingSink.foreach(_.publishSummary())

Compose the two in either order — wrap a SummarizingNotificationSink in a FailureOnlyNotificationSink to page only on a failed check or a job summary, or wrap a FailureOnlyNotificationSink in a SummarizingNotificationSink if the tallies themselves should reflect only failures.

Retrying any other sink in code

HttpNotificationSink retries on its own, configured by properties (above). Any other sink that reaches outside the JVM — KafkaNotificationSink, or one you wrote — can fail for reasons that have nothing to do with your job: a timeout, a momentary outage on the receiving end. RetryingNotificationSink wraps such a sink in code to retry a failed publish a few times, with capped exponential backoff, before giving up — and if every attempt fails, it publishes the event to a dead-letter sink instead of letting it disappear:

import com.invaract.verification.notification.{
  FileNotificationSink, HttpNotificationSink, NotificationSinkFactory, RetryingNotificationSink
}

val configuredSink = NotificationSinkFactory.create(notifyConfig) // e.g. HttpNotificationSink

val deadLetter = new FileNotificationSink
deadLetter.configure(Map("path" -> "/var/log/invaract/dead-letter.jsonl"))

val retryingSink = configuredSink.map(new RetryingNotificationSink(_, deadLetter))

// ... pass retryingSink to the enforcement rule and the listener, same as any other
// Option[NotificationSink] ...

The dead-letter sink is any ordinary NotificationSink — there’s no dedicated dead-letter type. Pointing a plain FileNotificationSink (or HadoopFsNotificationSink, for S3/GCS/HDFS) at a separate path is enough to get a durable, replayable record of whatever never made it through.

Defaults and tuning

maxAttempts defaults to 3, initialBackoffMs to 100, backoffMultiplier to 2.0 — so a persistently-failing sink is tried at roughly t=0, t=100ms, t=300ms before the event is dead-lettered. All three are constructor arguments: new RetryingNotificationSink(delegate, deadLetter, maxAttempts = 5, initialBackoffMs = 200L, backoffMultiplier = 3.0).

Retries are synchronous, not a background queue

Each retry blocks whatever thread called publish (Spark’s own analysis or listener thread) for the backoff duration. A bounded number of short retries keeps this predictable; there’s no scheduled retry-from-dead-letter after that — a sink down for longer than a few short retries is treated as down for a while, and the dead-lettered events are there for you to replay by hand once it’s back.

Composes with the two sinks above like any other wrapper — retry around a summarizing sink, or summarize around a retrying one, depending on whether the retry/dead-letter behavior should apply to the raw per-event traffic or only to published summaries.

The Kafka sink is a separate module

Unlike the three sinks above, KafkaNotificationSink isn’t part of spark-adapter — it ships as its own module, notification-kafka/, with a real dependency on kafka-clients. This is deliberate: most users of the verification engine don’t need Kafka, and kafka-clients is a real, sizeable dependency that would otherwise land on every user’s classpath whether they use it or not (the same reasoning behind Delta, Iceberg, Hive, Avro, and ClickHouse all being handled without spark-adapter itself depending on their client libraries — see Connector Support).

To use it, build the module’s own jar and add it alongside spark-adapter’s:

cd notification-kafka && sbt assembly

Then point sink.class at it, with sink.property.topic naming the destination topic. Every other sink.property.* key is passed straight through as a Kafka producer config — sink.property.bootstrap.servers, sink.property.security.protocol, and so on, exactly as Kafka’s own producer configuration reference documents them:

sink.enabled=true
sink.class=com.invaract.verification.notification.kafka.KafkaNotificationSink
sink.property.topic=invaract-events
sink.property.bootstrap.servers=localhost:9092

A user who doesn't want Kafka never resolves kafka-clients

spark-adapter’s own build.sbt declares no dependency on kafka-clients at all — not even a test-scoped one. Only building notification-kafka itself pulls it in.

Writing your own sink

A real destination — a message queue, an HTTP endpoint, a metrics system — is ordinary code implementing NotificationSink:

import com.invaract.verification.notification.{NotificationEvent, NotificationSink}

class MySink extends NotificationSink {
  private var endpoint: String = _

  override def configure(properties: Map[String, String]): Unit =
    endpoint = properties("endpoint")

  override def publish(event: NotificationEvent): Unit = {
    // event is one of ContractValidationEvent, WriteEvent, ContractInferenceEvent or
    // DryRunSummaryEvent (dry-run reporting), or (if you're behind a
    // SummarizingNotificationSink) JobSummaryEvent - send it however your
    // destination expects.
  }

  // Optional. A sink that delivers asynchronously can wait here, for at most
  // timeoutMs, for what it has already accepted - called once at application end.
  override def flush(timeoutMs: Long): Unit = ()
}

Point sink.class at it (it needs a public no-arg constructor — configuration happens separately, via configure) and it’s picked up the same way the built-in sinks are.

A broken sink can't break your job

Every sink returned by NotificationSinkFactory.create is wrapped so that an exception from publish — a network timeout, a bug in your sink — is caught and logged, never propagated. Notification is a best-effort side channel; it can never turn an otherwise successful contract check or write into a job failure. Construction-time misconfiguration (the previous step) is the only thing that throws.

What’s next

See View Verification Results for the other way to inspect what Invaract decided, and Prove Enforcement with the Regression Pack for how this repository’s own CI proves notification events are published on both the PASS and FAIL path against a real Spark job.