Message brokers and stream-processing ingestion

Many applications already publish durable domain events to a broker. A narrow consumer can normalize those events and deliver them to GraphJSON without putting analytics network calls in the customer request.

GraphJSON does not currently provide managed Kafka, Amazon SQS, Kinesis, Google Pub/Sub, or RabbitMQ connectors. Run the consumer in infrastructure you control and use the GraphJSON bulk logging endpoint as its destination.

application transaction
  → outbox or broker
  → analytics consumer
  → validate and minimize
  → batch up to 50 events per collection
  → GraphJSON
  → checkpoint only after accepted delivery

This guide specializes the general Reliable event delivery pattern for broker semantics.

Decide which stream to consume#

Prefer a stream of stable business facts:

• account created
• payment succeeded
• order fulfilled
• workflow completed
• job failed terminally

Avoid consuming raw database mutations when the business meaning is unclear. Use Database change data capture when row changes are the intended source.

Record:

• topic, queue, or subscription owner
• schema and compatibility policy
• partition or ordering key
• expected rate and burst
• retention and replay window
• source checkpoint authority
• sensitive fields to remove

Normalize one stable envelope#

Convert provider-specific metadata into a portable internal record:

{
  "source": "kafka",
  "stream": "billing.domain_events",
  "partition": 12,
  "offset": 918441,
  "source_event_id": "evt_01J...",
  "occurred_at": "2026-07-25T18:20:00Z",
  "schema_name": "payment_succeeded",
  "schema_version": 3,
  "body": {
    "event": "payment_succeeded",
    "account_id": "acct_7",
    "amount": 4900,
    "currency": "usd"
  }
}

The GraphJSON event should contain only the analytical body plus bounded lineage fields required for reconciliation. Do not copy broker credentials, headers, tracing baggage, full source payloads, or dead-letter error messages.

Preserve the original occurrence timestamp when calling GraphJSON.

Understand the delivery contract#

Most broker consumers should provide at-least-once processing:

1. receive one or more messages
2. validate and transform them
3. send them to GraphJSON
4. confirm a successful HTTP response
5. commit or acknowledge the source messages

If the worker crashes after GraphJSON accepts the batch but before the source checkpoint commits, the messages can be delivered again. Use stable event_id values and deduplicate important metrics in SQL. GraphJSON ingestion does not enforce event-ID uniqueness.

Do not checkpoint before GraphJSON accepts the events; that turns a temporary destination failure into permanent loss.

Batch by destination collection#

The bulk endpoint accepts at most 50 events in one collection. Group a received set by its destination collection, then create batches of 50 or fewer.

received broker messages
  → validate
  → group by GraphJSON collection
  → chunk to ≤ 50
  → send with bounded concurrency
  → checkpoint only accepted source positions

Do not combine unrelated sensitivity or retention classes merely to fill a batch. Collection architecture remains a governance boundary.

Start below the documented workspace guardrail of 40 ingestion requests per second. Use bounded concurrency, exponential backoff with jitter, and a circuit breaker when GraphJSON or the network is unhealthy.

Preserve ordering only where required#

Global ordering is rarely available or necessary. Define the smallest key that needs ordering:

• subscription ID for billing transitions
• order ID for order lifecycle
• account ID for lifecycle state
• logical job ID for attempts

Kafka and Kinesis preserve ordering within a partition or shard. SQS standard queues and ordinary Pub/Sub delivery can reorder messages. RabbitMQ ordering can be affected by multiple consumers and requeueing.

If order matters, choose a compatible source key and process that key serially, or make every event independently interpretable with effective_at, sequence, and version fields.

Do not rely on arrival order in GraphJSON to reconstruct authoritative state.

Checkpoint safely#

Kafka#

Commit offsets only after every event through that partition offset has been accepted or deliberately quarantined. Do not commit the highest completed offset when an earlier offset is still retrying.

SQS#

Set visibility timeout longer than the expected processing time and extend it for bounded retries. Delete a message only after successful GraphJSON delivery. Use FIFO queues and a message-group key only when the use case needs ordering.

Kinesis#

Checkpoint per shard after accepted records. A poison record must not block a shard indefinitely; quarantine it with its sequence number and schema failure.

Pub/Sub#

Ack after destination acceptance. Size the ack deadline and flow-control settings so the worker does not lease more messages than it can safely process.

RabbitMQ#

Use manual acknowledgments and a bounded prefetch. Reject permanent schema failures to a dead-letter exchange rather than requeueing forever.

Classify failures#

Keep retry budgets finite. A permanent message must not block all later partitions or consume unlimited broker retention.

Quarantine without leaking data#

A dead-letter record should include:

{
  "source_event_id": "evt_01J...",
  "stream": "billing.domain_events",
  "partition": "12",
  "position": "918441",
  "failure_class": "unknown_schema_version",
  "observed_schema_version": 7,
  "consumer_version": "analytics-consumer-4.2.0",
  "first_failed_at": "2026-07-25T18:21:00Z",
  "attempts": 1
}

Store the full payload only in an access-controlled source system when required. Do not put secrets or customer content in alerts.

Handle schema evolution#

Validate before transformation. Support additive compatible changes where safe, reject changed types or semantics, and keep the source schema version in the analytical event.

Deploy consumers in this order:

1. add support for the new compatible source version
2. test fixtures and a staging topic
3. deploy consumers
4. begin publishing the new version
5. monitor both versions
6. retire old support after the replay window closes

Use Executable event contracts for schemas and fixtures.

Replay deliberately#

Give every replay a bounded identity and interval:

source stream
partition or shard range
start and end positions
event-time interval
schema versions
consumer version
destination collection
expected counts

Throttle replay separately from live traffic. Prioritize current events when a large replay would delay operational dashboards.

Stable event IDs make duplicates identifiable, but a replay can still change metrics if its transformation logic changed. Version the mapping and reconcile before release.

Monitor the whole path#

Track:

• source backlog or consumer lag
• oldest unprocessed event age
• messages received, transformed, delivered, retried, and quarantined
• GraphJSON batch response classes
• batch size and request rate
• source-to-GraphJSON visibility delay
• missing event IDs and schema versions
• checkpoint progress by partition or shard

Monitor the broker and worker outside GraphJSON. A broken analytics consumer cannot reliably report its own outage only through its destination.

Reconcile independently#

For settled windows compare:

1. eligible messages published by the source
2. positions consumed
3. valid events transformed
4. events accepted by GraphJSON
5. unique event IDs visible in GraphJSON
6. quarantined and intentionally excluded records

Counts alone are insufficient when duplicates are possible. Compare IDs or partition-position ranges for critical streams.

Production checklist#

• The source stream contains stable, interpretable facts.
• Original event time and source event ID are preserved.
• Messages acknowledge only after GraphJSON accepts them.
• Batches contain at most 50 events in one collection.
• Concurrency stays within the workspace ingestion guardrail.
• Ordering requirements are explicit and scoped to one key.
• Retries distinguish temporary and permanent failures.
• Dead letters contain bounded metadata, not sensitive payloads.
• Schema rollout and replay procedures are versioned.
• Broker lag is monitored independently of GraphJSON.
• Settled IDs and positions reconcile across the complete path.