Cluster-Wide Ingest Streams¶

New in Quine Enterprise 2.1.1

Cluster-wide ingest streams are available from Quine Enterprise 2.1.1 onward. Upgrading past 2.1.1 requires a full cluster restart rather than a rolling, node-by-node upgrade. See the Release Notes.

RBAC Requirement

Managing cluster-wide ingest streams requires the same roles as ordinary ingest streams: creating, planning, and updating requires Architect or DataEngineer; pausing and resuming requires Architect, DataEngineer, or SRE; deleting requires Architect or DataEngineer; viewing requires Analyst, SRE, Architect, or DataEngineer. See OIDC and RBAC Setup for details.

Note that the dry run endpoint (:plan) requires write permission, not merely read. A dry run creates nothing, but it submits a configuration, credentials included, and makes the cluster connect to the source system with it.

Overview¶

An ordinary ingest stream runs on exactly one cluster member. That member opens the connection, reads every record, and runs the ingest query. If you want a second member reading the same Kafka topic, you define a second ingest stream by hand, and you divide the work yourself.

A cluster-wide ingest stream is one definition that the cluster splits for you. You submit a single ingest configuration; the cluster cuts the source into slices and decides which host runs each one.

The two mechanisms are independent. Ordinary ingest streams are unchanged, they use different API endpoints, and the two can run side by side in the same graph.

	Ingest stream	Cluster-wide ingest stream
Endpoint	`/ingests`	`/clusterIngests`
Runs on	One member you choose	Hosts the cluster chooses, which need not serve the graph and may each run several slices
Splitting the source	You define one stream per piece	Derived from the source
Moving the work	Delete and recreate	The cluster relocates it
Availability	Quine and Quine Enterprise	Quine Enterprise only

Cluster-wide ingest preserves the at-least-once delivery guarantee. Idempotent ingest queries remain the user-level contract, and they matter more here than for a single-member ingest, because slices relocate as a matter of routine.

Slices¶

Quine divides a cluster-wide ingest into slices and decides how many. The count is not something you request. Each slice is an ordinary ingest configuration with its source narrowed to one piece, and it runs on the same ingest pipeline as a single-member ingest stream. Slices keep their own resume point, and the API identifies each one by sliceIndex.

How far a source divides depends on what the source itself supports. Where a source has no natural number of its own, the default is four.

Source	How Quine divides it
Kafka, topic subscription	One per broker partition, sharing a consumer group; the broker balances them
Kafka, explicit partition assignments	One per named partition
Kinesis KCL	One per open shard, sharing an `applicationName`; the DynamoDB lease table balances them
SQS	Four interchangeable consumers; the queue's visibility timeout balances them
Number iterator, bounded	Contiguous sub-ranges, up to four, capped by the record count
Number iterator, unbounded	Not divided
S3	Not divided
Delta Sharing CDF	Not divided
Server-sent events, WebSocket, Reactive Stream	Not divided, and moving one overlaps its old and new hosts
File, Standard input	Not divided¹
Kinesis (SDK)	Refused
WebSocket file upload	Refused

A refused source cannot run cluster-wide at all, and Why a configuration is refused gives the reason. For the rest of the per-source detail, see the individual source pages: Kafka, Kinesis, SQS, Files and Named Pipes, Standard Input, and Reactive Streams.

Where slices run¶

The cluster decides where each slice runs, and may move it. Placement is not something you set, and it is not stored, so the only answer to "where is this slice?" is where it is running now. The runningOn field in the status response gives you that.

Slices run on hosts, a wider set than the members serving the graph, and they prefer hosts that hold no graph position. Such a host is called an ingest executor, and it appears in slice status with "tier": "ingestExecutor" and a null memberIdx. See Cluster-Wide Ingest Placement for what that means when sizing a cluster.

Defining a cluster-wide ingest stream¶

The request body wraps an ordinary ingest configuration:

{
  "ingest": {
    "name": "numbers",
    "source": {
      "type": "NumberIterator",
      "startOffset": 0,
      "limit": 1000
    },
    "query": "MATCH (n) WHERE id(n) = idFrom($that) SET n.num = $that",
    "onStreamError": { "type": "LogStreamError" },
    "maxPerSecond": 100
  }
}

Field	Description
`ingest`	An ordinary ingest configuration, exactly as you would submit it to `/ingests`. Every source type, format, record decoder, transformation, and error handling option behaves the same way.
`pinnedTo`	Optional. Pins the ingest to one host, given as `address:port` using the host's cluster address. See Pinning an ingest to one host.

Rate and concurrency limits¶

maxPerSecond and parallelism mean what they mean for any ingest stream: the records per second this ingest may process, and the records it may work on at once on one member. Dividing the source into slices changes neither.

To hold the whole ingest to maxPerSecond, each slice is given a share of it, so an ingest limited to 100 records per second and divided into four slices runs each slice at 25. This appears in warnings, since each slice's running configuration then carries a number you did not submit.

Planning before you create¶

Run the dry run first. It validates the configuration and returns the slice split it would deploy, without creating anything:

POST /api/v2/graph/{graphName}/clusterIngests:plan

Two reasons this is worth doing:

Planning consults the source system for partition counts and shard counts, so how many slices a configuration becomes is not something the configuration alone reveals.
A configuration that cannot be deployed cluster-wide at all is refused here, in exactly the terms a create would refuse it.

A plan response:

{
  "name": "numbers",
  "namespace": "quine",
  "sliceCount": 4,
  "slices": [
    { "sliceIndex": 0 },
    { "sliceIndex": 1 },
    { "sliceIndex": 2 },
    { "sliceIndex": 3 }
  ],
  "notes": [
    "maxPerSecond 100 is the whole ingest's budget: each of the 4 slices runs at 25/s"
  ]
}

The plan does not return the derived per-slice configurations, which would repeat your credentials back to you. sliceCount tells you how many pieces the source divides into, and notes lists every way the plan differs from what you submitted.

Configuration derived on your behalf¶

For some sources the plan supplies settings your configuration left out, because a cluster-wide ingest needs them and a single-member ingest does not. Every derivation is reported in warnings, since the running configuration now says something your submitted one did not.

Source	Derived	Why
Kafka	`groupId`, defaulted to the ingest's name	Slice workers are named `name#0`, `name#1`, and so on. Without a shared group, each would form its own consumer group and consume everything, giving you N-fold duplication.
Kafka	`offsetCommitting` set to `ExplicitCommit`	Without committed offsets, every worker restart, relocation, or consumer group rebalance re-reads from `autoOffsetReset`, and slices rebalance far more often than a single-member ingest restarts.
Kafka	`partition.assignment.strategy` set to `CooperativeStickyAssignor`	The client default rebalances eagerly, revoking every partition from every consumer on any change. A cluster-wide ingest rebalances for reasons a single-member ingest never has, and under the eager protocol each of those would stop the whole ingest rather than only the partitions that changed hands.

If you set any of these yourself, your value is kept.

Why a configuration is refused¶

Refusals are answered as 400, at plan time and at create time alike. Every one of them means that accepting the ingest would lose or misplace data. Nothing is refused for want of capacity, and nothing is refused because a source system was slow to answer.

Refusal	Cause
Member-local source	A local file or standard input, which a host chosen by the cluster may not have. Supply `pinnedTo` to name the host that does.
Connection-bound source	A WebSocket file upload arrives over a client connection that terminates at whichever host the client reached. Pinning does not route the connection, so a pinned slice would run and read nothing.
Kafka auto-commit enabled	`enable.auto.commit=true` commits offsets for records whose ingest query has not completed, so a rebalance or restart silently drops them.
Kinesis SDK source	The SDK source carries no resume state, and its default `Latest` iterator type makes a relocated slice restart at the stream's present, skipping everything that arrived while the slice moved. Use the KCL source, whose DynamoDB lease table is exactly the resume state this one lacks.
Pinned multi-slice	A pin was given for a configuration that plans into more than one slice. One pin cannot say where a second slice belongs, and pinning the first while scattering the rest would be a placement nobody chose.
Non-positive `maxPerSecond` or `limit`	Unlimited is the field being absent, so a value at or below zero can only be a mistake.

You cannot tell from a configuration alone how many slices it becomes, since planning asks the source system for partition and shard counts. Use the dry run to see the split before you create anything.

A worked example¶

Plan, create, watch, and remove one cluster-wide ingest. The source is a bounded number iterator, so it divides into four slices and finishes on its own.

1. Plan it. Nothing is created, and the response tells you how the source divides.

curl -X "POST" "http://127.0.0.1:8080/api/v2/graph/quine/clusterIngests:plan" \
     -H 'Content-Type: application/json' \
     -d '{
  "ingest": {
    "name": "numbers",
    "source": { "type": "NumberIterator", "startOffset": 0, "limit": 1000 },
    "query": "MATCH (n) WHERE id(n) = idFrom($that) SET n.num = $that",
    "maxPerSecond": 100
  }
}'

2. Create it. The same body, to the collection endpoint. A 201 carries the record, including any warnings.

curl -X "POST" "http://127.0.0.1:8080/api/v2/graph/quine/clusterIngests" \
     -H 'Content-Type: application/json' \
     -d '{
  "ingest": {
    "name": "numbers",
    "source": { "type": "NumberIterator", "startOffset": 0, "limit": 1000 },
    "query": "MATCH (n) WHERE id(n) = idFrom($that) SET n.num = $that",
    "maxPerSecond": 100
  }
}'

3. Watch it. Only the single-ingest read carries sliceStatus, so this is where you see per-slice progress and which host holds each slice.

curl "http://127.0.0.1:8080/api/v2/graph/quine/clusterIngests/numbers"

Check sliceStatus.status first. running while it works, completed once every slice has read its share. If it reads degraded, slicesNotReporting names the slices that did not answer.

4. Pause and resume it. Both apply to every slice at once, and both keep each slice's resume point.

curl -X "POST" "http://127.0.0.1:8080/api/v2/graph/quine/clusterIngests/numbers:pause"
curl -X "POST" "http://127.0.0.1:8080/api/v2/graph/quine/clusterIngests/numbers:resume"

5. Remove it.

curl -X "DELETE" "http://127.0.0.1:8080/api/v2/graph/quine/clusterIngests/numbers"

Pinning an ingest to one host¶

pinnedTo names one host as address:port, using that host's cluster address, the same value reported as runningOn in slice status.

{
  "ingest": {
    "name": "audit-log",
    "source": {
      "type": "File",
      "path": "file_ingests/audit.log",
      "format": {
        "type": "Line"
      }
    },
    "query": "MATCH (n) WHERE id(n) = idFrom($that) SET n.line = $that"
  },
  "pinnedTo": "10.0.0.7:2551"
}

A pin means two different things depending on the source.

For a local file or standard input the pin is required. Without it these sources are refused, because the cluster has no way to know which hosts hold the input. Pinning is how you run one under cluster management: the definition belongs to the cluster rather than to a single member, it survives that host restarting, and it is managed through the same API as any other cluster-wide ingest. It does not divide into more than one slice.

For any other source a pin is a preference, and dropping it returns the slice to the cluster's choice.

A pinned ingest is never relocated. It runs on its host or nowhere, and it stops until that host returns. A mistyped address looks identical from the outside, so a pin naming a host the cluster does not know is reported in warnings when you create it, and listed in pinnedAwaitingHost in status for as long as it stays unmatched.

Updating in place¶

PATCH redefines a running cluster-wide ingest stream. The cluster rebuilds each slice worker from the new definition, keeping the resume point it has been writing. Delete-and-recreate starts every slice over instead.

You can change:

the ingest query
maxPerSecond
parallelism
error handling: onRecordError, onStreamError, dead letter queue settings

You cannot change:

Change	Result
The source	`400`. Each slice's resume point is recorded against the piece of the source it was given. Delete and recreate instead.
The number of slices	`400`. Same reason. If the source has changed shape and you want the new split, delete and recreate.
The name	`400`. Resume state is keyed by name. A body naming a different ingest than the path is refused rather than treated as a rename.

An update replaces pinnedTo

pinnedTo is part of the definition and is replaced along with it. An update that omits the field unpins the ingest, making its slice relocatable. This is reported in warnings, but it is easy to do by accident when you meant only to change the query. The exception is a source whose pin is load-bearing, such as a local file or standard input, where dropping the pin is refused rather than allowed to hand the slice to a host without the input.

Pausing, resuming, and deleting¶

:pause and :resume apply to the whole ingest, every slice on every host. The goal field reports what you asked for, and sliceStatus.status reports what the slices are doing. When those disagree the ingest reads as degraded.

Deleting removes the definition, and the cluster stops every slice worker running under it.

Pausing and resuming leave the generation unchanged. Redefining the ingest changes it, because generation is a content fingerprint of the definition, config and slices together.

Error handling¶

Record-level and stream-level error handling, retry settings, and dead letter queues are configured exactly as for an ordinary ingest stream, on the ingest object. Each slice worker applies them independently, and dead letter queue destinations are written from whichever member is running the slice at the time.

Two failure modes have no single-member equivalent.

A poisoned ingest. When a slice fails in a way the cluster will not retry past, the whole ingest is marked failed and stops being placed. Status reports "status": "failed", and the failure field carries the reason. To recover, redefine the ingest with PATCH, or delete and recreate it.

A stalled slice. A slice that has stopped for a reason the cluster will keep retrying reports "stalled", and its detail field names the cause. Unlike failed, a stalled slice has not been given up on.

Inspecting cluster-wide ingest streams via the API¶

Operation	Endpoint
List all cluster-wide ingest streams	List Cluster-Wide Ingest Streams: `GET /api/v2/graph/{graphName}/clusterIngests`
Create a cluster-wide ingest stream	Create Cluster-Wide Ingest Stream: `POST /api/v2/graph/{graphName}/clusterIngests`
Plan a cluster-wide ingest stream	Plan a Cluster-Wide Ingest Stream: `POST /api/v2/graph/{graphName}/clusterIngests:plan`
Get status	Cluster-Wide Ingest Stream Status: `GET /api/v2/graph/{graphName}/clusterIngests/{clusterIngestName}`
Update in place	Update Cluster-Wide Ingest Stream: `PATCH /api/v2/graph/{graphName}/clusterIngests/{clusterIngestName}`
Pause	Pause Cluster-Wide Ingest Stream: `POST /api/v2/graph/{graphName}/clusterIngests/{clusterIngestName}:pause`
Resume	Resume Cluster-Wide Ingest Stream: `POST /api/v2/graph/{graphName}/clusterIngests/{clusterIngestName}:resume`
Delete	Delete Cluster-Wide Ingest Stream: `DELETE /api/v2/graph/{graphName}/clusterIngests/{clusterIngestName}`

Cluster-wide ingest streams are managed through the API only. The UI displays them but does not create, pause, or resume them, and it shows no per-slice detail: a cluster-wide ingest appears as one row alongside ordinary ingest streams, with its throughput summed across every member running a slice of it. Slice-level status is available only from the single-ingest read described below.

The Dashboard additionally lists poisoned cluster-wide ingest streams, which run nowhere and so produce none of the live telemetry the other rows are built from.

The ingest record¶

Every response that carries a cluster-wide ingest returns its definition and current disposition:

Field	Description
`name`	The ingest's name.
`namespace`	The graph it ingests into.
`generation`	A content fingerprint of the definition, config and slices together. Changes when the ingest is redefined; a pause or resume leaves it unchanged.
`goal`	The disposition you set: `running` or `paused`.
`slices`	The slices the ingest is cut into, each with a `sliceIndex` and its `pinnedTo`, if any. Where each one is running is in `sliceStatus`, not here, because the cluster decides that and can change it.
`failure`	Why this ingest is poisoned, if it is. Present on every response carrying a record, including list responses.
`warnings`	Ways the deployed plan differs from what you requested. Present on create and update responses.
`sliceStatus`	Live aggregate status. Only present on single-ingest status requests.

Reading `sliceStatus`¶

sliceStatus is the live picture, gathered from the hosts when you ask for it.

Field	Description
`goal`	The disposition you set: `running` or `paused`.
`status`	What the slices are actually doing: `running`, `paused`, `degraded`, `stalled`, `failed`, or `completed`.
`slicesPlaced`	How many slices the record defines.
`slicesReporting`	How many distinct slices answered at the current generation. Never exceeds `slicesPlaced`.
`slicesNotReporting`	The slice indexes that did not answer. Listed rather than left to be inferred, because a placed slice that never reports is the failure an aggregate hides.
`slicesReportedByMultipleHosts`	Slice indexes claimed by two hosts at the current generation. Expected and brief during a relocation of a source that overlaps; anything longer is worth investigating. The counters alone cannot show this, since `ingestedCount` takes the furthest report per slice.
`pinnedAwaitingHost`	Pinned slices whose host is not currently in the cluster, each with its `sliceIndex` and `pinnedTo`. Empty whenever every pin matches a known host.
`ingestedCount`	Records ingested, summed across slices. Within a slice the furthest report wins, since two reports for one slice are two measurements of the same quantity.
`bytesIngested`	Bytes ingested, summed across every report.
`recordsPerSecond`, `bytesPerSecond`	Current rates, summed across every report. During an overlap both hosts really are consuming.
`startTimeMillis`	When the earliest slice started.
`totalRuntimeMillis`	The longest slice's runtime.
`failure`	Present when the ingest is poisoned.
`slices`	Per-slice detail, ordered by slice then host.

Each entry in slices:

Field	Description
`sliceIndex`	Which slice this is.
`runningOn`	The host answering for it: `address`, `port`, and `uid`, plus `memberIdx` and `tier`. `uid` is the host's actor system incarnation, so two reports an hour apart with the same address are the same machine, and the same `uid` makes them the same process.
`runningOn.tier`	`positioned` for a host serving the graph, `ingestExecutor` for a host running slices while holding no graph position, or `unknownToThisView` for a host that answered but was absent from the membership view the gather was attributed against. `memberIdx` is null for the latter two.
`status`	`Running`, `Paused`, `Blocked`, `Stalled`, a terminal engine status such as `Completed` or `Failed`, or `StatusUnavailable` when the host could not answer for this slice in time.
`detail`	Why the slice is in that status, when the status alone does not say. Set for `Stalled`, where the cause is the thing to go and fix.
`engine`	Which internal execution path the slice worker was built with: `classic`, `channel`, `channel-local`, `scattered-channel`, or `none` when no worker is running. Diagnostic detail, not something to configure or depend on.
`generation`	The generation this report was made against.
`superseded`	`true` for a predecessor still draining under a generation the record has moved past. Shown rather than dropped, because its counters are still climbing and its host is still busy, but it does not count toward coverage.
`ingestedCount`, `bytesIngested`, `recordsPerSecond`, `bytesPerSecond`, `totalRuntimeMillis`	This slice's own meters.

Reading a degraded ingest

An ingest is degraded for one of two reasons. Either a slice is not reporting, in which case slicesNotReporting names it, and pinnedAwaitingHost says whether a pin explains the absence. Or every slice is reporting but one of them disagrees with the goal: a slice that is Blocked, a slice that is Paused while the goal is running, or a slice still running while the goal is paused. The per-slice status field distinguishes these.

slicesReportedByMultipleHosts is not a cause of degraded. Two hosts answering for one slice is counted once toward coverage, and it is expected while a source that overlaps on relocation changes hands.

Next steps¶

Cluster-Wide Ingest Placement: how slices are placed, relocated, and recovered
Ingest Streams: record formats, decoders, transformations, and error handling, all of which apply unchanged
Delivery Guarantees: what at-least-once means when slices move

A file or standard-input ingest reads from one host's filesystem or console, so it needs one extra field naming that host. See Pinning an ingest to one host. ↩