Kafka ingest: commit offset only after the TRoE write succeeds#1968
Open
cfrey-de wants to merge 11 commits into
Open
Kafka ingest: commit offset only after the TRoE write succeeds#1968cfrey-de wants to merge 11 commits into
cfrey-de wants to merge 11 commits into
Conversation
The Kafka consumer committed the offset in kafkaConsumerLoop() right after kafkaBatchProcess() returned - but that function only performs the MongoDB (current-state) upsert. The TRoE (temporal/Postgres) write happens afterwards, in requestCompleted(), via serviceP->troeRoutine(). So the ordering was: 1. MongoDB upsert 2. Kafka offset commit <-- point of no return 3. TRoE / Postgres write If step 3 failed (Postgres down, connection drop, SQL error) the temporal instance was lost permanently: MongoDB held the newer value, the history did not, and the offset was already committed so the batch was never redelivered. Fix: - pgCommands() now records TRoE write failures in orionldState.troeError. It also catches SQL-level command failures via PQresultStatus() (constraint violation, deadlock, aborted transaction, ...) - previously only a NULL result or a dropped connection was detected, so a failed INSERT that left the connection alive slipped through and was only noticed (if at all) at COMMIT time. - kafkaConsumerLoop() now runs requestCompleted() (which performs the TRoE write) BEFORE committing, and commits the offset only when both the MongoDB upsert and the TRoE write succeeded (mongoOk && !orionldState.troeError). A failed TRoE write leaves the offset uncommitted, so Kafka redelivers the batch - turning "in the topic" into a real durability boundary. The orionldState.troeError field (previously marked "Unused - TODO: remove") is repurposed for this signal. REST ingest is unaffected by this change: there the TRoE write already runs in requestCompleted() after the response is sent, which is a separate, more invasive problem (it would require moving the TRoE write into the request path to fail the response on TRoE error). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
CLA Assistant Lite bot All contributors have signed the CLA ✍️ |
cfrey-de
force-pushed
the
fix/kafka-troe-durability-before-offset-commit
branch
2 times, most recently
from
June 30, 2026 06:33
a70dc50 to
8662fab
Compare
Resolve pgCommands.cpp conflict: keep develop's richer SQL diagnostics (execStatus + PQresultErrorMessage + SQL in the message) AND the branch's orionldState.troeError durability flag at every failure exit. develop improved the error REPORTING; this branch adds the durability SIGNAL - complementary. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
New -kafkaAckTopic (empty = disabled). When set, the Kafka consumer path emits
per-batch feedback to that topic so a producer can confirm true end-to-end delivery
and dead-letter poison the offset-commit model would otherwise lose silently:
- ACK (lean: count/offsets/batchIds) when a batch is durable in MongoDB AND TRoE.
- NACK (with the failed entities) when the Mongo upsert or TRoE write failed - the
payload is carried because mongoOk is batch-level and a malformed entity may be
recoverable nowhere; committing a later batch would skip its offset.
- NACK (raw bytes) for a message that could not even be parsed.
Kafka path only (kafkaConsumerLoop.cpp) - REST ingest keeps its synchronous HTTP
response as its acknowledgement; nothing added to shared pgCommands()/requestCompleted().
New producer module kafkaAckProducer.{h,cpp} (broker was consumer-only). CLI var
kafkaAckTopic declared in orionldState.h like the other config vars. C-style only (no STL).
NOTE: not build-verified in the authoring env (kjson/K-lib headers absent).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Separate test (troe_kafka_ack_nack): a valid entity produces an ACK on the -kafkaAckTopic; a malformed message produces a parse-fail NACK carrying the raw bytes. DISABLED like the existing Kafka test (needs a running broker). The TRoE-failure NACK shares the same kafkaAckSend() path and is not reproduced here (forcing a live TRoE failure mid-run is environment-dependent). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
cfrey-de
marked this pull request as draft
July 3, 2026 07:58
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| // every part at full length plus the fixed scaffolding, so the bounded snprintf/escape never | ||
| // overrun and 'pos' stays < bufSize. | ||
| // | ||
| size_t entitiesLen = (entitiesJson != NULL) ? strlen(entitiesJson) : 0; |
| // overrun and 'pos' stays < bufSize. | ||
| // | ||
| size_t entitiesLen = (entitiesJson != NULL) ? strlen(entitiesJson) : 0; | ||
| size_t errorLen = (error != NULL) ? strlen(error) : 0; |
| // | ||
| size_t entitiesLen = (entitiesJson != NULL) ? strlen(entitiesJson) : 0; | ||
| size_t errorLen = (error != NULL) ? strlen(error) : 0; | ||
| int bufSize = (int) (256 + strlen(offsetsJson) + strlen(batchIds) + entitiesLen + errorLen * 2); |
| // | ||
| size_t entitiesLen = (entitiesJson != NULL) ? strlen(entitiesJson) : 0; | ||
| size_t errorLen = (error != NULL) ? strlen(error) : 0; | ||
| int bufSize = (int) (256 + strlen(offsetsJson) + strlen(batchIds) + entitiesLen + errorLen * 2); |
The Kafka consumer uses auto.offset.reset=latest, so the valid entity produced right after brokerStart was skipped (consumer not yet assigned its partitions) - never persisted, no ACK, so the ACK assertion timed out. Produce the valid entity in a retry loop until it lands (produceUntilPersisted). Also make waitForAck use a wall-clock deadline instead of counting iterations (each ackDump blocks ~4s, so the old 'elapsed++' loop ran ~5x its intended timeout - hence the ~400s runtime). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
cfrey-de
marked this pull request as ready for review
July 3, 2026 11:09
…lowed) pgDatabaseTableCreateAll() ran the whole schema DDL via PQexec() but only checked for a NULL result. A failed statement (e.g. a GEOGRAPHY column when the postgres server has no postgis extension) aborts the transaction yet returns a non-NULL result, so the failure slipped through: the broker committed the aborted transaction and ran on with NO TRoE tables and nothing logged - a misconfigured Postgres meant silent temporal-data loss. Now the result status is checked (same fix as pgCommands), so it rolls back and logs the SQL error loudly. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…a category
A batch-write NACK previously sent only a fixed category ('TRoE (Postgres) write failed' /
'MongoDB upsert failed'), so a downstream consumer (e.g. the DataGrabber dead-letter) could
not tell WHY - a missing-schema error looked identical to a constraint violation or a
connection drop.
- orionldState gains troeErrorString[512], cleared each cycle by the existing bzero().
- pgCommands() stashes the real libpq text at every failure site: PQresultErrorMessage() for
a failed statement, PQerrorMessage() for connection failures, trimmed of the trailing
newline. Also resolves the old 'FIXME: string! (last error?)'.
- kafkaConsumerLoop appends it -> 'TRoE (Postgres) write failed: <pg error>', and for the
Mongo path appends orionldState.pd.detail when present.
kafkaAckSend() already sizes its buffer from the actual error length (errorLen*2 for escaping),
so the longer string is sent without truncation. The parse-fail NACK is unchanged, and the
(disabled) ack/nack functional test only asserts 'status:nack', so it is unaffected.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
| snprintf(orionldState.troeErrorString, sizeof(orionldState.troeErrorString), "%s", text); | ||
|
|
||
| // libpq error strings end in a newline; trim trailing whitespace so the NACK reads cleanly | ||
| int last = (int) strlen(orionldState.troeErrorString) - 1; |
The stricter PQresultStatus() check added earlier made pgDatabaseTableCreateAll() fatal on ANY schema-creation error - including the benign 'duplicate object' that the non-idempotent CREATE TYPEs raise when the TRoE schema is already there (a broker restart against a populated db, or the functional-test harness, which pre-creates the schema via pgCreate). That regressed broker startup with 'type valuetype already exists'. Now duplicate_object / duplicate_table / duplicate_schema / duplicate_function (SQLSTATE 42710 / 42P07 / 42P06 / 42723) are treated as 'schema already present - reuse it' (matching the original code's intent and the header note), while every other failure - notably a missing postgis GEOGRAPHY type - still aborts startup loudly. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…st core only) The REST path sets orionldState.in.contentType from the HTTP Content-Type, so batchEntitiesFinalCheck() resolves each entity's @context and expands its type and attribute terms against it. The Kafka ingest path (kafkaBatchProcess) set up orionldState for the batch upsert but never set contentType, so it stayed at the core-context default: an inline @context in a Kafka entity was ignored and terms expanded against the core/default context (e.g. type "Cycle" became default-context/Cycle instead of the producer's vocabulary term). Mirror the REST behaviour: when the batch carries an inline @context, flag it as JSON-LD (MT_JSONLD) so the existing per-entity @context handling in batchEntitiesFinalCheck() applies the producer's context. Batches without an inline @context keep the previous JSON/core-context behaviour (non-breaking).
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Kafka ingest: TRoE-durable offset commit + optional end-to-end ACK/NACK feedback
1. Commit the Kafka offset only after the TRoE write succeeds
The Kafka consumer committed the offset in
kafkaConsumerLoop()right afterkafkaBatchProcess()returned - but that function only performs the MongoDB (current-state) upsert. The TRoE (temporal/Postgres) write happens afterwards, inrequestCompleted(), viaserviceP->troeRoutine(). So the ordering was:If step 3 failed (Postgres down, connection drop, SQL error) the temporal instance was lost permanently: MongoDB held the newer value, the history did not, and the offset was already committed so the batch was never redelivered.
Fix:
pgCommands()now records TRoE write failures inorionldState.troeError. It also catches SQL-level command failures viaPQresultStatus()(constraint violation, deadlock, aborted transaction, ...) - previously only a NULL result or a dropped connection was detected, so a failed INSERT that left the connection alive slipped through and was only noticed (if at all) at COMMIT time.kafkaConsumerLoop()now runsrequestCompleted()(which performs the TRoE write) BEFORE committing, and commits the offset only when both the MongoDB upsert and the TRoE write succeeded (mongoOk && !orionldState.troeError). A failed TRoE write leaves the offset uncommitted, so Kafka redelivers the batch - turning "in the topic" into a real durability boundary.The
orionldState.troeErrorfield (previously marked "Unused - TODO: remove") is repurposed for this signal. REST ingest is unaffected by this change: there the TRoE write already runs inrequestCompleted()after the response is sent, which is a separate, more invasive problem (it would require moving the TRoE write into the request path to fail the response on TRoE error).2. Optional end-to-end ACK/NACK feedback (
-kafkaAckTopic)The durability fix makes "in the topic" a real boundary for the consumer, but a Kafka producer still only sees the broker ack - it cannot tell whether Orion actually persisted the batch (MongoDB and TRoE). And because the consumer commits the whole current position, a message the broker accepts but that Orion cannot process (a poison / malformed entity) is skipped as soon as a later batch commits - silently lost and invisible to the producer.
When
-kafkaAckTopic <topic>is set, the Kafka consumer publishes one feedback record per processed batch to that topic:Each record carries the batch's
{partition, offset}and echoes any producer-suppliedx-batch-idheader, so a producer can correlate the feedback to what it sent: confirm true end-to-end delivery (advance its own watermark only once Orion holds the data) and dead-letter poison instead of losing it.Opt-in and Kafka-path-only: an empty
-kafkaAckTopickeeps the current behaviour, and REST ingest is unaffected (its synchronous HTTP response is its acknowledgement). A newkafkaAckProducermodule owns the feedback producer (the broker was consumer-only until now); nothing is added to the sharedpgCommands()/requestCompleted()path.Record format:
{"status":"ack|nack","count":N,"offsets":[{"partition":P,"offset":O}],"batchIds":[...],"error":"...","entities":[...]}(An ACK omits
errorandentities.)Tests
test/functionalTest/cases/0000_ld/troe/troe_kafka_ack_nack.test- a valid entity produces an ACK; a malformed message produces a parse-fail NACK carrying the raw bytes. Disabled by default (like the existingtroe_kafka_consumer_ingestion.test) as it needs a running Kafka broker; the header documents the KRaft setup.