diff --git a/docs/assets/images/admin/onlinefs/offsets_card.png b/docs/assets/images/admin/onlinefs/offsets_card.png new file mode 100644 index 0000000000..451a75a817 Binary files /dev/null and b/docs/assets/images/admin/onlinefs/offsets_card.png differ diff --git a/docs/assets/images/admin/onlinefs/offsets_topic.png b/docs/assets/images/admin/onlinefs/offsets_topic.png new file mode 100644 index 0000000000..314e83c62e Binary files /dev/null and b/docs/assets/images/admin/onlinefs/offsets_topic.png differ diff --git a/docs/setup_installation/admin/ha-dr/dr.md b/docs/setup_installation/admin/ha-dr/dr.md index 7336af9694..b07c3c8425 100644 --- a/docs/setup_installation/admin/ha-dr/dr.md +++ b/docs/setup_installation/admin/ha-dr/dr.md @@ -394,7 +394,23 @@ olk: indices: "-myindex" ``` -### In-Place Restore +#### OnlineFS offsets after the restore { #restore-onlinefs-offsets } + +The RonDB backup holds the Kafka offsets OnlineFS saved, in `onlinefs.kafka_offsets`, each with the ID of the topic it was saved for. +Kafka on the new cluster starts without the topics of the backed-up cluster, creates each online topic again when it is first written to, and gives it a new ID. +OnlineFS reads each partition of a recreated topic from its start instead of resuming at the restored offset, so the offsets need no step after the restore, and no record written to the new topics is skipped. + +- Until OnlineFS saves an offset for a recreated topic, [Kafka offsets][onlinefs-offsets-admin] in `Cluster Settings` → `OnlineFS` shows the restored offsets as `Earlier topic`, or as `Topic missing` while Kafka has not created the topic yet. +- After an [in-place restore][in-place-restore], Kafka keeps its topics and their IDs, and OnlineFS resumes at the restored offsets: it writes the records that arrived since the backup again, which brings the restored online feature store up to date with them. +- Each OnlineFS pod runs its `offsets-schema` step when it starts, which leaves the restored table as it is. + If the step fails, the pod does not start, as described in [If the offsets-schema step fails][upgrade-onlinefs-offsets-failures]. +- On a Kafka that gives its topics no IDs, OnlineFS cannot tell a recreated topic from the earlier one and resumes at the restored offsets. + On such a Kafka, stop OnlineFS after the restore, delete the rows of `onlinefs.kafka_offsets` and start OnlineFS again, so that it reads every online topic from its start. +- Offset changes that were pending at the backup come back with `onlinefs.kafka_offset_changes`. + OnlineFS rejects them on a new cluster, since each was asked for a topic Kafka no longer has. + After an in-place restore, one restored within 24 hours of its request applies when OnlineFS next gets the partition; cancel it in [Kafka offsets][onlinefs-offsets-admin] if it should not. + +### In-Place Restore { #in-place-restore } !!! Note In-place restore is available from Hopsworks version 4.8.0. diff --git a/docs/setup_installation/admin/onlinefs.md b/docs/setup_installation/admin/onlinefs.md index f7f12e5e6e..d937f3600d 100644 --- a/docs/setup_installation/admin/onlinefs.md +++ b/docs/setup_installation/admin/onlinefs.md @@ -264,8 +264,12 @@ Administrators also see the revision of the saved settings, the revision the las ## Consumer offsets { #onlinefs-offsets } -OnlineFS stores its consumer offsets in the online feature store of each project, per consumer group, topic and partition. -It keeps the offsets of a saved connection apart from those of the Helm connection, and starts new ones whenever the saved `bootstrap.servers` changes: +OnlineFS stores its consumer offsets in the `onlinefs.kafka_offsets` table of the RonDB it writes to, per consumer group, topic and partition, with the ID Kafka gives the topic the offset was saved for. +When a topic is deleted and created again under the same name, after a restore that rebuilt Kafka or by hand, Kafka gives it a new ID, and OnlineFS reads each partition of the new topic from its start instead of resuming at an offset of the earlier topic. +A running OnlineFS sees the new topic at its next Kafka metadata refresh, every 60 seconds with the chart's `metadata.max.age.ms`. +If a topic is deleted and created again between two refreshes, OnlineFS may keep its position on the new topic until it restarts, and then reads the new topic from its start. + +OnlineFS keeps the offsets of a saved connection apart from those of the Helm connection, and starts new ones whenever the saved `bootstrap.servers` changes: - With the Helm connection, OnlineFS uses the offsets it has always used, so an upgraded cluster resumes where it was. - A saved connection whose `bootstrap.servers` differs from the previous one, or a connection saved where there was none, starts a new set of offsets. @@ -279,6 +283,86 @@ A new broker list for the same cluster, or moving a connection from the Helm val Plan such a change for a time when that load is acceptable. A change of the Helm connection itself, in the Helm values, keeps the offsets of the Helm connection, as in earlier releases. +### Kafka offsets { #onlinefs-offsets-admin } + +The `Kafka offsets` card lists the topics OnlineFS consumes: for each project with an online feature group, every topic its online feature groups write to, the project's online topic or one a feature group names itself. +Each row shows the project, the topic, its number of partitions in Kafka and the offset changes pending for it. +`Read only` marks a topic whose offsets cannot be changed, with the reason in its tooltip, as listed in [Changing offsets][onlinefs-offset-changes]. +The search matches the project or the topic name. + +
+ The Kafka offsets card listing two topics of the fraud_detection project with their partition counts and pending changes +
The topics OnlineFS consumes
+
+ +A click on a topic opens its offsets, for each consumer group and partition: the offset OnlineFS resumes from, the offsets Kafka still has, from the first one to the end offset, the lag, and a status. +Every partition Kafka has is listed, also one OnlineFS saved no offset for. + +
+ The offsets of the fraud_detection_onlinefs topic for the RONDB consumer group, with the actions to change them and an applied change that moved partition 3 to its start +
A topic's offsets and its offset changes
+
+ +| Status | Meaning | +| --- | --- | +| `Current topic` | The offset was saved for the topic Kafka has under this name now. | +| `Earlier topic` | The offset was saved for an earlier topic of the same name, so OnlineFS reads the partition from its start. | +| `No topic ID` | The offset was saved without a topic ID, or Kafka gives its topics none, so which topic it belongs to is not known. | +| `Topic missing` | Kafka has no topic of this name. | +| `Kafka unavailable` | Hopsworks could not ask Kafka, so the offsets Kafka has and the lag are unknown. | +| `Not saved` | OnlineFS saved no offset for the partition, so it reads it from its start. | + +OnlineFS consumes each topic in two consumer groups: `RONDB`, which writes the online feature store, and `VECTORDB`, which writes embeddings to the vector database. +With a saved Kafka connection the groups carry a number after an `@`, such as `RONDB@3`, which changes with the saved `bootstrap.servers` as described above. +A consumer group marked `Not read` belongs to an earlier Kafka connection, and OnlineFS no longer reads its offsets. +`RONDB` is always listed, and `VECTORDB` once it saved an offset for the topic. +OnlineFS runs consumers of a group only while its thread number is above 0: with `service.vectorDbThreadNumber` at `0`, the default without VectorDB ingestion, the `VECTORDB` offsets stay as they were. + +The lag is the number of records OnlineFS has still to process, written or not. +It shows for the current topic, and for a partition with no saved offset, in a consumer group of the current connection; elsewhere the page shows `-`, never 0. +The offsets are those in the RonDB of the region the Hopsworks backend runs in. + +#### Changing offsets { #onlinefs-offset-changes } + +In a topic's offsets, an administrator can move partitions of one consumer group: + +- `Set offset` moves one selected partition to a given offset, from the first offset Kafka still has up to the end offset. +- `Move to start` moves the selected partitions, or all of them, to the first offset Kafka still has, to process their records again. +- `Move to end` moves them to the end offset, to skip every record written until then. + +The confirmation lists each partition with its saved offset, the offsets Kafka has and the new offset. +Hopsworks then stores a change for each partition in `onlinefs.kafka_offset_changes` and restarts OnlineFS, as described in [Applying changes][onlinefs-applying-changes]. +OnlineFS applies a change when it next gets the partition, normally during that restart: it saves the new offset and the outcome of the change together, and resumes the partition from it. + +- Every OnlineFS pod restarts, one after another, so ingestion of every topic slows or pauses until the rollout ends. +- The `RONDB` and `VECTORDB` groups move separately, each with changes of its own. +- Moving back processes the records again under the current rules: rows are written and deletes applied again, ingestion notifications repeat, and readers can see older values until OnlineFS catches up. + Records Kafka no longer retains cannot be read again. +- Moving forward skips records, which never reach the online feature store. +- The offline feature store is not affected. + +| Change state | Meaning | +| --- | --- | +| `Pending` | Stored, and waiting for OnlineFS to get the partition. | +| `Applied` | OnlineFS saved the new offset. The change shows the offset before and after. | +| `Rejected` | OnlineFS refused it and kept the saved offset, with the reason: the topic was deleted and created again after the change was asked for, Kafka gives the topic no ID, or the offset is no longer in the partition's range, as when Kafka deleted older records meanwhile. | +| `Expired` | Not applied within 24 hours of the request. | +| `Cancelled` | Cancelled with `Cancel` while pending. | +| `Replaced` | A later change for the same partition and consumer group was asked for while it was pending. | + +A partition has at most one pending change per consumer group: asking for another replaces it, also when two administrators ask at the same time. +A change stays pending while OnlineFS does not get the partition, as for a group whose thread number is 0 or when OnlineFS was not restarted. +When changes are pending and the [status][onlinefs-status] shows no rollout, press `Restart`. +Each time OnlineFS gets partitions, it gives Kafka 30 seconds for the topic IDs and the ranges of the partitions with changes; a change whose range Kafka does not give in that time stays pending until the next time. +Every change is kept, with who asked for it and when, its outcome, and the offsets before and after. + +A topic is read only, and its changes can be neither asked for nor cancelled, when: + +- the project's online feature store is on a satellite, whose OnlineFS this cluster does not restart; +- the OnlineFS that runs does not apply changes, as an OnlineFS older than 5.3 creates no `onlinefs.kafka_offset_changes`; +- Kafka could not be asked, has no topic of this name, or gives the topic no ID; +- the saved [subscription][onlinefs-subscription] leaves the topic out; a subscription in the Helm values is not checked. + ## Startup { #onlinefs-startup } Each OnlineFS pod asks Hopsworks for the saved settings once at startup, before it consumes. @@ -350,3 +434,45 @@ All three restart OnlineFS as described in [Applying changes][onlinefs-applying- Every logged-in user can read the status of OnlineFS with `GET /hopsworks-api/api/onlinefs/status`, also with an API key of scope `FEATURESTORE`. It returns the state, the replica counts, the message, the settings state, and whether any setting is saved, without the revisions and the restart error. + +### Kafka offsets { #onlinefs-offsets-rest-api } + +Administrators can read and change the [Kafka offsets][onlinefs-offsets-admin], also with an API key of scope `ADMIN`: + +```text +GET /hopsworks-api/api/admin/onlinefs/offsets?filter=&offset=&limit= +GET /hopsworks-api/api/admin/onlinefs/offsets/projects/{projectId}/topics/{topic} +POST /hopsworks-api/api/admin/onlinefs/offsets/projects/{projectId}/topics/{topic}/changes +DELETE /hopsworks-api/api/admin/onlinefs/offsets/projects/{projectId}/topics/{topic}/changes/{changeId} +``` + +- The list returns a page of `items`, by project and topic name, with `count`, the number of topics the filter matches; `limit` is at most 100. +- A topic returns its `topicId`, its `readOnlyReason` when its offsets cannot be changed, `items` with an entry per consumer group and partition, and its `changes`: every pending change and the 500 newest ended ones, latest request first. + A field left out is unknown, such as the `offset` of a partition OnlineFS saved none for. +- `status` is one of `CURRENT_TOPIC`, `EARLIER_TOPIC`, `TOPIC_ID_UNKNOWN`, `TOPIC_MISSING`, `KAFKA_UNAVAILABLE` and `NOT_SAVED`, the statuses of the table above in that order. + +A change moves partitions of one consumer group: + +```json +{ + "consumerGroup": "RONDB", + "partitions": [3], + "target": "OFFSET", + "offset": 1981, + "topicId": "qIquf00TSxOofDURgAgSlg" +} +``` + +- `target` is `OFFSET`, `EARLIEST` or `LATEST`, and `offset` is set for `OFFSET` only. +- An empty `partitions` moves every partition the topic has. +- `topicId` repeats the one the topic returned, so a change is refused once the topic was deleted and created again since. +- The answer holds the stored `changes`, with their `changeId`, and `restartRequested`, false when the restart could not be requested: the changes then stay pending until OnlineFS restarts. + +| Status | Error code | When | +| --- | --- | --- | +| 400 | `270378` | The change is invalid: no target, a consumer group OnlineFS does not read, an offset with `EARLIEST` or `LATEST`, a partition the topic does not have, or an offset outside the partition's range. | +| 404 | `270375` | No online feature group of the project writes to the topic. | +| 409 | `270376` | The topic is read only; the message gives the reason. | +| 409 | `270377` | The topic was deleted and created again since its offsets were read. | +| 409 | `270379` | The change to cancel is no longer pending. | +| 503 | `270374` | The offsets or the changes cannot be read or stored, as before OnlineFS has created its tables. When RonDB did not confirm the commit of a change, the message says it may have been stored: read the topic before asking again. | diff --git a/docs/setup_installation/admin/upgrade.md b/docs/setup_installation/admin/upgrade.md index ddac87ecf6..33fdc95c17 100644 --- a/docs/setup_installation/admin/upgrade.md +++ b/docs/setup_installation/admin/upgrade.md @@ -460,6 +460,83 @@ Clients keep their endpoint, Knative-mode deployments stay on Istio, and no depl - Every Standard-mode deployment answers through the usual endpoint, with both URL shapes. - The Knative-mode deployments are unchanged. +### OnlineFS offsets move to `onlinefs.kafka_offsets` { #upgrade-onlinefs-offsets } + +From 5.3, OnlineFS keeps its consumer offsets in one table, `onlinefs.kafka_offsets`, in the RonDB it writes to, with the ID of the Kafka topic each offset was saved for, in place of a `kafka_offsets` table in the online feature store of each project. +With the topic ID, OnlineFS reads a topic that was deleted and created again under the same name from its start, as after a restore that rebuilt Kafka, instead of resuming at an offset of the earlier topic ([Consumer offsets][onlinefs-offsets]). + +- **No step before `helm upgrade`.** + Each OnlineFS pod runs an `offsets-schema` init container before OnlineFS starts. + It creates the table, copies the offsets of every project into it, and then drops the project tables. + For each consumer group, topic and partition it keeps the highest offset the project tables hold, with the topic's current ID, and leaves out the offsets of topics Kafka no longer has. + A pod that starts later finds nothing to move. + It also creates `onlinefs.kafka_offset_changes`, which holds the offset changes administrators ask for in [Kafka offsets][onlinefs-offsets-admin]. +- **It connects to the DDL SQL node of the release as `hopsworksroot`**: the `ddl-mysqld` Service of the release, or, when RonDB is managed externally on a release with `global._hopsworks.full_platform: true`, the consul name in `onlinefs.dependencies.mysql.ddlConsulServiceName` (`mysqlddl`). + A satellite cluster (`global._hopsworks.full_platform: false`) always connects to the Service, also with an external RonDB. + A cluster that disables the DDL SQL node (`rondb.rondb.meta.ddlMySQLd.enabled: false`) sets `onlinefs.dependencies.mysql.serviceName: mysqld`, and also `onlinefs.dependencies.mysql.ddlConsulServiceName: mysql` when it is not a satellite and RonDB is managed externally. + `onlinefs.dependencies.mysql.url` overrides both. +- **A step that fails keeps the pod from starting.** + It reads every project table before it writes, and drops none before the copy is committed, so a step that fails while it copies leaves every project table as it is. + One that fails while it drops the tables has copied every offset already, and the next start drops the rest. + The pod stays in `Init:Error` or `Init:CrashLoopBackOff`, and the log names the database: + + ```sh + kubectl logs -c offsets-schema -n + ``` + + The step gives up after 15 minutes, its wait for Hopsworks to answer with the OnlineFS settings included, and Kubernetes starts it again. + Fix the cause, as described in [If the offsets-schema step fails][upgrade-onlinefs-offsets-failures], and the next start of the pod runs it again. +- **Records written during the rollout can be written to the online feature store twice, never skipped.** + An OnlineFS pod of the earlier release that still runs while a new pod moves the offsets saves to project tables that are gone, and its partitions resume from the moved offsets. +- **On satellite clusters**, upgraded after central as [Strimzi 1.3][upgrade-strimzi-13] requires, a project created after central's upgrade has no offsets table on a satellite that is not upgraded yet. + The satellite's OnlineFS logs `Failed to save offsets` for that project and keeps writing its rows, but reads the project's online topics again from their start when it restarts, and once more at the satellite's upgrade. +- **Rolling OnlineFS back to 5.2** makes it read every online topic again from its earliest retained record, because the project tables it reads are gone. + An offset change still pending stays in `onlinefs.kafka_offset_changes` unapplied; once OnlineFS is upgraded again, a change more than 24 hours old expires instead of applying. +- **A restore needs no step for the offsets on a Kafka that gives its topics IDs**, as the Kafka the chart installs does. + `onlinefs.kafka_offsets` is in the RonDB backup like any other database, and after a restore that rebuilt Kafka, OnlineFS reads the recreated topics from their start. + An offset saved without a topic ID, by a consumer group of an earlier Kafka connection or on a Kafka that gives no IDs, has nothing to check against, and OnlineFS resumes from it. + +#### If the offsets-schema step fails { #upgrade-onlinefs-offsets-failures } + +A step that fails logs `Failed to set up onlinefs.kafka_offsets, OnlineFS does not start`, followed by the error, in the `offsets-schema` container of a pod that is not `Running`: + +```sh +kubectl logs -c offsets-schema -n +``` + +| The error | What it means | What to do | +| --- | --- | --- | +| `Hopsworks did not answer with the OnlineFS settings within 900 s` | Hopsworks did not answer within the step's 15 minutes. Nothing was changed. | Wait until the `hopsworks-instance` pods are ready. The next start runs the step again. | +| `Communications link failure`, `UnknownHostException` or `Access denied for user 'hopsworksroot'` | The step cannot reach or log in to the SQL node. Nothing was changed. | While the upgrade restarts the SQL node, this passes on its own. Otherwise check the SQL node the step connects to, as described above, and the `hopsworksroot` key of the `mysql-users-secrets` Secret. | +| `Got error 4009 'No data node(s) available'` | RonDB's data nodes are restarting or down, as while the upgrade restarts them. Nothing was changed. | The next start runs the step again once the data nodes are up. | +| `onlinefs.kafka_offsets has the columns [...], OnlineFS maps [...]` | The table exists with columns OnlineFS does not map, because it was created or changed outside OnlineFS. Nothing was moved. | Do not drop it, it holds the offsets. Change its columns to the ones the message lists after `OnlineFS maps`, keeping its rows. | +| `onlinefs.kafka_offset_changes has the columns [...], OnlineFS maps [...]` | The table of offset changes exists with columns OnlineFS does not map. Nothing was moved. | It holds only the changes administrators asked for and their outcomes. Drop it, and the next start creates it again. | +| `Failed to read a per-project kafka_offsets, it is kept as it is`, with the database | The table of that project cannot be read. A single such table stops the whole step: nothing was written, and every project table is kept. | Repair that table, from the RonDB backup if needed. A table that cannot be repaired can be dropped, as described below. | +| A Kafka error, such as `org.apache.kafka.common.errors.TimeoutException` | The step could not ask Kafka for the IDs of the topics. Nothing was written. | Check that OnlineFS reaches Kafka, as for its own consumers. | +| `onlinefs.kafka_offsets was not set up within 900 s` | The step ran out of time while it moved the offsets. | The next start goes on. Offsets it moved already keep their values. | +| The error of a `DROP TABLE` statement | Every offset is in `onlinefs.kafka_offsets`, and the project tables before the failing one are dropped. | Fix the cause. The next start drops the remaining tables. | + +The commands below run SQL as the RonDB `root` user on the chart's `mysqlds-0` pod; with an external RonDB, run the statements on one of its SQL nodes instead. +A project table that cannot be repaired can be dropped, so that the step moves the others: + +```sh +NAMESPACE=hopsworks +ROOT_PW=$(kubectl get secret mysql-users-secrets -n "$NAMESPACE" -o jsonpath='{.data.root}' | base64 -d) +kubectl exec mysqlds-0 -n "$NAMESPACE" -c mysqld -- env MYSQL_PWD="$ROOT_PW" mysql -uroot -e \ + 'DROP TABLE ``.kafka_offsets' +``` + +OnlineFS then has no offsets for the online topics of that project and reads them from their earliest retained record, so it writes the records Kafka still retains to the online feature store again, and skips none. + +The step has finished when every OnlineFS pod is `Running`, the `offsets-schema` log ends with `Moved the per-project kafka_offsets into onlinefs.kafka_offsets` or `No per-project kafka_offsets left to move`, and no project table is left: + +```sh +kubectl exec mysqlds-0 -n "$NAMESPACE" -c mysqld -- env MYSQL_PWD="$ROOT_PW" mysql -uroot -N -e \ + "SELECT table_schema FROM information_schema.tables WHERE table_name = 'kafka_offsets' AND table_schema <> 'onlinefs'" +``` + +[Kafka offsets][onlinefs-offsets-admin] in `Cluster Settings` → `OnlineFS` then shows the moved offsets as `Current topic`. + ### Search index rebuild { #upgrade-53-search-reindex } After the upgrade, the `post-upgrade-opensearch-reindex-job` hook indexes the existing data sources, which 5.3 adds to search, and datasets, which a cluster upgraded from 5.1 may lack, without emptying the index. diff --git a/docs/user_guides/fs/feature_group/topic.md b/docs/user_guides/fs/feature_group/topic.md index f1986e382a..e112e0fa7f 100644 --- a/docs/user_guides/fs/feature_group/topic.md +++ b/docs/user_guides/fs/feature_group/topic.md @@ -103,6 +103,11 @@ Topics count against the project's Kafka topic quota, which an administrator can When the cluster is configured to use an [external Kafka cluster][external-kafka-cluster], Hopsworks creates the topic there too, unless it already exists, in which case it is used as it is. The `kafka_connector` data source of the project needs permission to create topics for that, as described in that guide. +## OnlineFS offsets { #onlinefs-project-offsets } + +OnlineFS saves, for each partition of an online topic, the offset it resumes from. +Cluster administrators see these offsets, and can move them, in `Cluster Settings` → `OnlineFS`, as described in [Kafka offsets][onlinefs-offsets-admin]. + ## API Reference [`FeatureGroup`][hsfs.feature_group.FeatureGroup]