> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need.

# Worker Events

Status: Experimental.

Workers can publish an event to [Kafka](https://kafka.apache.org) when a job starts extracting a table, as it
progresses, and when it completes or fails. Other systems can then react as it happens, instead of asking Amp
repeatedly with the [Job Status API](/managing-amp/jobs/status): a dashboard that shows how far each dataset has synced,
an alert when a job fails, or a log of what Amp did.

Events are off by default. When they are, or when Kafka is unavailable, the Job Status API still reports the same
progress.

## Prerequisites

* A Kafka cluster that your workers can reach, or a service compatible with it, like Amazon MSK, Confluent Cloud, or
  Redpanda.
* A topic for the events. Amp doesn't create it. Give it at least as many partitions as Amp's `partitions` setting,
  `16` by default:

  ```bash
  kafka-topics --create \
    --bootstrap-server localhost:9092 \
    --topic amp.worker.events \
    --partitions 16 \
    --replication-factor 3
  ```

## Turn events on

Add the `[worker.events]` and `[worker.events.kafka]` sections to the workers' configuration, then restart them:

```toml [config.toml]
[worker.events]
enabled = true

[worker.events.kafka]
brokers = ["kafka-1:9092", "kafka-2:9092"]
topic = "amp.worker.events"
partitions = 16
```

When a worker connects to Kafka, it logs `Kafka event streaming enabled`. When it can't, within 30 seconds, it logs
`Failed to create Kafka producer, events disabled`, and runs its jobs without events until it restarts.

To watch the events arrive:

```bash
kafka-console-consumer --bootstrap-server localhost:9092 --topic amp.worker.events --from-beginning
```

The events are binary [Protocol Buffers](/running-amp/worker-events#event-format), so the console shows them as raw bytes. Your consumer decodes
them.

### Authentication and TLS

Managed Kafka services usually require SASL authentication over TLS:

```toml [config.toml]
[worker.events.kafka]
brokers = ["pkc-12345.us-east-1.aws.confluent.cloud:9092"]
topic = "amp.worker.events"
sasl_mechanism = "SCRAM-SHA-256"
tls_enabled = true
```

Keep the credentials out of the file, in environment variables:

```bash
export AMP_CONFIG_WORKER__EVENTS__KAFKA__SASL_USERNAME="$KAFKA_USERNAME"
export AMP_CONFIG_WORKER__EVENTS__KAFKA__SASL_PASSWORD="$KAFKA_PASSWORD"
```

When your brokers use a certificate signed by your own certificate authority, set `tls_ca_cert_path` to its PEM file.

## Settings

| Setting                                                                                                          | Default               | Description                                                                                        |
| ---------------------------------------------------------------------------------------------------------------- | --------------------- | -------------------------------------------------------------------------------------------------- |
| [`worker.events.enabled`](/running-amp/configuration-reference#workereventsenabled)                              | `false`               | Publishes events.                                                                                  |
| [`worker.events.progress_interval`](/running-amp/configuration-reference#workereventsprogress_interval)          | `10`                  | The least time between two progress events for the same table, in seconds.                         |
| [`worker.events.event_queue_capacity`](/running-amp/configuration-reference#workereventsevent_queue_capacity)    | `1024`                | How many events a worker holds while it waits to send them. Beyond that, it drops new ones.        |
| [`worker.events.kafka.brokers`](/running-amp/configuration-reference#workereventskafkabrokers)                   | none, required        | The Kafka brokers' addresses, like `["kafka-1:9092"]`.                                             |
| [`worker.events.kafka.topic`](/running-amp/configuration-reference#workereventskafkatopic)                       | `"amp.worker.events"` | The topic to publish to.                                                                           |
| [`worker.events.kafka.partitions`](/running-amp/configuration-reference#workereventskafkapartitions)             | `16`                  | How many of the topic's partitions Amp spreads events across. At most the topic's partition count. |
| [`worker.events.kafka.sasl_mechanism`](/running-amp/configuration-reference#workereventskafkasasl_mechanism)     | none                  | `"PLAIN"`, `"SCRAM-SHA-256"`, or `"SCRAM-SHA-512"`. Unset, Amp doesn't authenticate.               |
| [`worker.events.kafka.sasl_username`](/running-amp/configuration-reference#workereventskafkasasl_username)       | none                  | The SASL username. Required with `sasl_mechanism`.                                                 |
| [`worker.events.kafka.sasl_password`](/running-amp/configuration-reference#workereventskafkasasl_password)       | none                  | The SASL password. Required with `sasl_mechanism`.                                                 |
| [`worker.events.kafka.tls_enabled`](/running-amp/configuration-reference#workereventskafkatls_enabled)           | `false`               | Connects to the brokers over TLS.                                                                  |
| [`worker.events.kafka.tls_ca_cert_path`](/running-amp/configuration-reference#workereventskafkatls_ca_cert_path) | none                  | A PEM file with the certificate authority to trust, instead of the system's.                       |

`progress_interval` is also how long an [export job](/managing-amp/pipelines) waits, by default, before it records its
progress at the destination when it finds no new rows, whether events are on or not. An export job's
`state_heartbeat_secs` overrides it: see [Catching Up](/managing-amp/pipelines/catching-up).

## Events

Each event is about one table of one job:

| Event            | When                                                                                                                  |
| ---------------- | --------------------------------------------------------------------------------------------------------------------- |
| `sync.started`   | The job starts extracting the table.                                                                                  |
| `sync.progress`  | The table has made progress since the last event, at most every `progress_interval` seconds.                          |
| `sync.completed` | The job reached its end block. A job without one, which follows the chain's head, never completes, so never sends it. |
| `sync.failed`    | The job failed with an error.                                                                                         |

A progress event describes where the table is now, not what changed since the last one, so receiving it twice is
harmless.

## Delivery

* **Order**: Amp publishes all events of a table to the same partition, so a consumer reads them in order. Each Kafka
  message's key is the table's `namespace/name/manifest_hash/table_name`.
* **Duplicates**: a consumer may receive an event more than once. Skip the ones whose `event_id` you've already seen.
* **Losses**: events never slow down or fail a job, so Amp doesn't guarantee their delivery. A worker drops an event
  when sending it fails or takes more than 30 seconds, or when `event_queue_capacity` events are already waiting, and
  logs a warning: `failed to send event to Kafka (event dropped)` or `event queue full; dropping event`. Use the
  [Job Status API](/managing-amp/jobs/status) to catch up after a gap.

## Event format

Each Kafka message's value is an `amp.job.events.v1.JobEvent` message, encoded with [Protocol
Buffers](https://protobuf.dev). Generate a decoder for your language from Amp's schema,
[`crates/job-core/proto/events.proto`](https://github.com/edgeandnode/amp/blob/f27102fd59608258dc58a6dbbaac880f124364ae/crates/job-core/proto/events.proto).

Every event has the same envelope, shown here as JSON:

```json
{
  "event_id": "019a3602-4b7c-7e3a-9f51-2c8d1e0b6a47",
  "event_type": "sync.progress",
  "event_version": "1.1",
  "timestamp": "2026-10-30T17:04:43.473520+00:00",
  "source": { "worker_id": "worker-01" },
  "sync_progress": { … }
}
```

| Field              | Description                                                                                        |
| ------------------ | -------------------------------------------------------------------------------------------------- |
| `event_id`         | A unique ID. Its first bits are the time, so IDs sort in the order Amp created the events.         |
| `event_type`       | `sync.started`, `sync.progress`, `sync.completed`, or `sync.failed`.                               |
| `event_version`    | The format's version, `1.1`. Version `1.0` events lack `job_kind`.                                 |
| `timestamp`        | When the worker created the event, in RFC 3339 format.                                             |
| `source.worker_id` | The ID of the worker that sent it.                                                                 |
| The payload        | One of `sync_started`, `sync_progress`, `sync_completed`, or `sync_failed`, matching `event_type`. |

Every payload says which job and table it's about:

| Field        | Description                                                                                                                    |
| ------------ | ------------------------------------------------------------------------------------------------------------------------------ |
| `job_id`     | The [job](/managing-amp/jobs)'s ID.                                                                                            |
| `job_kind`   | `materialize-raw`, `materialize-derived`, or `materialize-phaser` for datasets, `adbc` for [exports](/managing-amp/pipelines). |
| `dataset`    | The dataset's `namespace`, `name`, and `manifest_hash`, the hash that identifies its [version](/datasets/versioning).          |
| `table_name` | The table's name.                                                                                                              |

Then each payload adds its own fields.

### `sync_started`

| Field         | Description                                                                           |
| ------------- | ------------------------------------------------------------------------------------- |
| `start_block` | The block the job starts from, when it has one.                                       |
| `end_block`   | The block the job stops at, when it has one. Jobs that follow the chain's head don't. |

### `sync_progress`

The `progress` field:

```json
{
  "job_id": 12345,
  "job_kind": "materialize-raw",
  "dataset": { "namespace": "acme", "name": "eth_mainnet", "manifest_hash": "2dbf16e8…" },
  "table_name": "blocks",
  "progress": {
    "start_block": 24000000,
    "current_block": 24000099,
    "network": "mainnet",
    "latest_segment_delay_seconds": 12,
    "files_count": 1,
    "total_size_bytes": 64768
  }
}
```

| Field                          | Description                                                                                                               |
| ------------------------------ | ------------------------------------------------------------------------------------------------------------------------- |
| `start_block`                  | The first block the table holds.                                                                                          |
| `current_block`                | The latest block the table holds.                                                                                         |
| `end_block`                    | The block the job stops at, when it has one.                                                                              |
| `network`                      | The table's network.                                                                                                      |
| `latest_segment_delay_seconds` | How far behind the present the table's latest block is, in seconds: how fresh it is. Unset when blocks have no timestamp. |
| `files_count`                  | How many files the table has.                                                                                             |
| `total_size_bytes`             | Their total size, in bytes.                                                                                               |

A table built from several networks has no single block range, so `start_block`, `current_block`, `end_block`, and
`network` are unset. Its file counts and freshness still tell you how it progresses.

### `sync_completed`

| Field             | Description                             |
| ----------------- | --------------------------------------- |
| `final_block`     | The last block the job extracted.       |
| `duration_millis` | How long the job took, in milliseconds. |

### `sync_failed`

| Field           | Description                           |
| --------------- | ------------------------------------- |
| `error_message` | The error that stopped the job.       |
| `error_type`    | The kind of error, when Amp knows it. |