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 partitions 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 partitions
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 partitions relocate as a matter of routine.

Partitions¶

Quine divides a cluster-wide ingest into partitions and decides how many. The count is not something you request. Each partition 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. Partitions keep their own resume point, and the API identifies each one by partitionIndex.

Two senses of the word

A Kafka topic and a Kinesis stream have partitions and shards of their own. Where the two need telling apart, this page says broker partition for Kafka's and partition for the unit Quine divides an ingest into. For a topic subscription they happen to correspond one-to-one.

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 broker 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 partitions run¶

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

Partitions 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. 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 partitions changes neither.

To hold the whole ingest to maxPerSecond, each partition is given a share of it, so an ingest limited to 100 records per second and divided into four partitions runs each partition at 25. This appears in warnings, since each partition'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 partition 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 broker partition counts and shard counts, so how many partitions 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",
  "partitionCount": 4,
  "partitions": [
    { "partitionIndex": 0 },
    { "partitionIndex": 1 },
    { "partitionIndex": 2 },
    { "partitionIndex": 3 }
  ],
  "notes": [
    "maxPerSecond 100 is the whole ingest's budget: each of the 4 partitions runs at 25/s"
  ]
}

The plan does not return the derived per-partition configurations, which would repeat your credentials back to you. partitionCount 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	Partition 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 partitions rebalance far more often than a single-member ingest restarts.
Kafka	`partition.assignment.strategy` set to `CooperativeStickyAssignor`	The client default rebalances eagerly, revoking every broker 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 broker 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 partition 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 partition restart at the stream's present, skipping everything that arrived while the partition moved. Use the KCL source, whose DynamoDB lease table is exactly the resume state this one lacks.
Pinned multi-partition	A pin was given for a configuration that plans into more than one partition. One pin cannot say where a second partition 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 partitions it becomes, since planning asks the source system for broker 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 partitions 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 status, so this is where you see per-partition progress and which host holds each partition.

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

Check status.state first. RUNNING while it works, COMPLETED once every partition has read its share. If it reads DEGRADED, status.partitionsNotReporting names the partitions that did not answer.

4. Pause and resume it. Both apply to every partition at once, and both keep each partition'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 partition 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 partition.

For any other source a pin is a preference, and dropping it returns the partition 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 partition worker from the new definition, keeping the resume point it has been writing. Delete-and-recreate starts every partition 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 partition's resume point is recorded against the piece of the source it was given. Delete and recreate instead.
The number of partitions	`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 partition 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 partition to a host without the input.

Pausing, resuming, and deleting¶

:pause and :resume apply to the whole ingest, every partition on every host. The goal field on the record reports what you asked for, and status.state reports what the partitions are doing. When those disagree the ingest reads as DEGRADED.

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

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 partition worker applies them independently, and dead letter queue destinations are written from whichever member is running the partition at the time.

Two failure modes have no single-member equivalent.

A poisoned ingest. When a partition fails in a way the cluster will not retry past, the whole ingest is marked failed and stops being placed. status.state reads FAILED, and the record's failure field carries the reason. To recover, redefine the ingest with PATCH, or delete and recreate it.

A stalled partition. A partition that has stopped for a reason the cluster will keep retrying reports STALLED, and its detail field names the cause. Unlike FAILED, a stalled partition is still being retried. The whole ingest reads STALLED while one of its partitions does, even when the others have finished, since the ingest has not yet read everything it was asked to read.

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. The UI displays them but does not create, pause, or resume them: a cluster-wide ingest appears as one row alongside ordinary ingest streams, with its throughput summed across every member running a partition of it. Per-partition detail comes 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.

A status read that no host answers returns 503. The condition is transient, so the distinct status code tells you to retry rather than read the answer as the ingest's current state. A single unreachable host is still reported in partitionsNotReporting.

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.
`settings`	The configuration this ingest is running, with secrets redacted: the same shape a per-member ingest reports, and the body a `PATCH` replays. Absent only if the stored configuration cannot be decoded.
`goal`	The disposition you set: `RUNNING` or `PAUSED`.
`partitions`	The partitions the ingest is cut into, each with a `partitionIndex` and its `pinnedTo`, if any. Where each one is running is reported in `status`, since the cluster decides that and can change it. `pinnedTo` requires `ClusterStatusRead`.
`status`	Live aggregate status. Only present on single-ingest status requests, because gathering it asks every host.
`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.

Reading `status`¶

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

Field	Description
`state`	What the ingest is actually doing: `RUNNING`, `PAUSED`, `DEGRADED`, `STALLED`, `COMPLETED`, or `FAILED`.
`partitionsPlaced`	How many partitions the ingest is cut into.
`partitionsReporting`	How many of those partitions answered this read.
`partitionsNotReporting`	The partition indexes that did not answer this read, either because the host running one did not answer in time or because it is not currently running anywhere. Each index is listed, so a partition needing attention is named rather than left to be inferred from a count.
`pinnedAwaitingHost`	Pinned partitions whose host is not currently in the cluster, each with its `partitionIndex` and `pinnedTo`. Empty whenever every pin matches a known host.
`ingestedCount`	Records the ingest has processed across every partition. See What the counts mean.
`bytesIngested`	Bytes processed, on the same terms as `ingestedCount`.
`recordsPerSecond`, `bytesPerSecond`	Current rates, summed across every partition.
`startTimeMillis`	When the ingest's first partition started.
`totalRuntimeMillis`	The longest-running partition's runtime.
`partitions`	Per-partition detail, ordered by partition index.

Each entry in status.partitions:

Field	Description
`partitionIndex`	Which partition this is.
`runningOn`	The host running it right now, as `address` and `port`. A relocatable partition can answer from a different host on a later read, so this reports where it is at this moment. Requires `ClusterStatusRead`; a caller with only `IngestRead` sees it blank.
`state`	What this partition is doing. See the table below.
`detail`	Why the partition is in that state, where the state alone does not say. Set for `STALLED` and `BLOCKED`, naming the condition to act on.
`ingestedCount`, `bytesIngested`	This partition's count on the host now running it, which restarts near zero when the partition moves to a host that has not run it before. See What the counts mean.
`recordsPerSecond`, `bytesPerSecond`, `totalRuntimeMillis`	This partition's own meters.

The partition states:

State	Meaning
`RUNNING`	Consuming.
`PAUSED`	Stopped because the ingest was paused, and will resume where it left off.
`RESTORED`	Rebuilt after its host restarted, and not yet consuming.
`BLOCKED`	Not consuming, though the ingest was not paused. `detail` names the cause: either the cluster is not operating, so ingest is held closed cluster-wide, or this worker alone is held closed. The first is resolved by restoring the cluster, the second by looking at this partition.
`STALLED`	Stopped and being retried; `detail` says what to fix.
`COMPLETED`	Read its whole share of the source.
`TERMINATED`	Stopped on purpose.
`FAILED`	Stopped by a failure while processing.
`UNKNOWN`	The host running this partition did not answer in time, so its state is not known. The partition is also listed in `partitionsNotReporting`.

What the counts mean¶

The ingest's ingestedCount and bytesIngested and the per-partition counts underneath answer different questions, so read each for what it measures rather than summing the second into the first.

The ingest's totals come from every member's persisted ingest ledger. That is what lets a count survive a partition moving between hosts: the work the old host did travels with the partition, and the total reads the same from whichever host answers. Members write that ledger about once a minute, so the total reads zero until the first write after the ingest starts and advances a minute at a time thereafter. Use it for how much this ingest has processed.

Each per-partition count is a live meter of one host's share of one partition. It is current to the moment, and it starts near zero on a host that has not run that partition before. Use it for what one host is doing right now.

Two entries at one partitionIndex

partitions can hold two entries with the same partitionIndex while a live source is being handed between hosts. Both are consuming, and each counts its own host's share, so neither is the partition's total. Briefly, this is expected; persistently, it is worth investigating.

Reading a degraded ingest

An ingest is DEGRADED for one of two reasons. Either a partition is not reporting, in which case partitionsNotReporting names it and pinnedAwaitingHost says whether a pin explains the absence. Or every partition is reporting but one of them disagrees with the goal: a partition that is BLOCKED or RESTORED, a partition that is PAUSED while the goal is RUNNING, or partitions still consuming while the goal is PAUSED. The per-partition state field distinguishes these.

Next steps¶

Cluster-Wide Ingest Placement: how partitions 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 partitions 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. ↩