Kafka ingest: commit offset only after the TRoE write succeeds

CHANGES_NEXT_RELEASE

@@ -1,4 +1,6 @@
 ## Fixed Issues
+  #XXXX - Kafka ingest: the consumer committed the Kafka offset right after the MongoDB upsert but BEFORE the TRoE (temporal/Postgres) write. A TRoE failure (Postgres down, connection drop, SQL error) therefore lost the temporal instance permanently - MongoDB kept the newer value, the history did not, and the offset was already gone so the batch was never redelivered. The offset is now committed only after BOTH the MongoDB upsert and the TRoE write succeed. pgCommands() now flags failures in orionldState.troeError (and additionally catches SQL-level command failures via PQresultStatus - previously only a dead connection or a NULL result was detected); a flagged batch is left uncommitted so Kafka redelivers it (cfrey-de - Carsten Frey)
+  #1943 - Fixed crash on a malformed q-filter with a missing AND/OR separator between two expressions (e.g. q=category=="office"category!="inactive" - missing ';'): qLexCheck did not validate the token following a String/Boolean/Integer/Float value, so the malformed token stream reached qParse and triggered a NULL dereference. Such input is now rejected with 400 Bad Request
   #1899 - Fixed broker startup stall on a cyclic @context in the persisted context cache: at startup the cached @contexts are reloaded, and an array-@context re-downloads its referenced contexts recursively. A context that references itself (directly or via a chain) made the loader wait the full download timeout (3s) per cycle in contextCacheWait - which exists to let one thread wait for ANOTHER thread already downloading the same context, but during the single-threaded startup recursion the "other downloader" is the same thread, so the wait could never be satisfied (and with enough cross-referencing contexts the cumulative stall trips orchestrator startup probes -> CrashLoop). The recursion now detects same-thread re-entry and rejects the cyclic @context immediately. Such non-fatal context-load failures are now logged as warnings instead of errors, so they no longer look like a failed startup
   #XXXX - Batch delete (non-legacy path): a duplicated Entity ID is now handled per NGSI-LD spec clause 10.3.6 - the first occurrence is deleted and each subsequent occurrence gets a ResourceNotFound (404) error in the 207 response (the duplicates were previously dropped silently)
   #1951 - Fixed slow, half-broken batch upsert of more than 1000 entities: the legacy "which of these already exist" lookup (mongoCppLegacyEntityListLookupWithIdTypeCreDate) truncated its result at a hardcoded 1000, so on a re-upsert the surplus entities were misclassified as new and re-inserted (duplicate-key path) instead of updated. The lookup's $in query is already bounded by the batch (and thus by the max payload size), so the arbitrary 1000 cap was removed

src/app/orionld/orionld.cpp

@@ -260,6 +260,7 @@
 char            kafkaBrokerList[512];
 char            kafkaTopic[256];
 char            kafkaGroupId[256];
+char            kafkaAckTopic[256];
 int             kafkaBatchSize       = 100;
 int             kafkaBatchLingerMs   = 50;
 int             kafkaConsumerThreads = 2;
@@ -375,6 +376,7 @@
 #define KAFKA_BROKER_DESC      "comma-separated list of Kafka broker addresses"
 #define KAFKA_TOPIC_DESC       "Kafka topic to consume NGSI-LD entities from"
 #define KAFKA_GROUP_DESC       "Kafka consumer group ID"
+#define KAFKA_ACK_TOPIC_DESC   "Kafka topic for TRoE ingest ACK/NACK feedback (empty = disabled)"
 #define KAFKA_BATCH_SIZE_DESC  "max entities per micro-batch before flush to database"
 #define KAFKA_LINGER_DESC      "max milliseconds to wait for a micro-batch to fill"
 #define KAFKA_THREADS_DESC     "number of Kafka consumer threads"
@@ -470,6 +472,7 @@
   { "-kafkaBrokerList",       kafkaBrokerList,          "KAFKA_BROKER_LIST",         PaString,  PaOpt,  _i "localhost:9092", PaNL, PaNL,            KAFKA_BROKER_DESC            },
   { "-kafkaTopic",            kafkaTopic,               "KAFKA_TOPIC",               PaString,  PaOpt,  _i "orionld-entities", PaNL, PaNL,          KAFKA_TOPIC_DESC             },
   { "-kafkaGroupId",          kafkaGroupId,             "KAFKA_GROUP_ID",            PaString,  PaOpt,  _i "orionld-consumer", PaNL, PaNL,           KAFKA_GROUP_DESC             },
+  { "-kafkaAckTopic",         kafkaAckTopic,            "KAFKA_ACK_TOPIC",           PaString,  PaOpt,  _i "",               PaNL, PaNL,            KAFKA_ACK_TOPIC_DESC         },
   { "-kafkaBatchSize",        &kafkaBatchSize,          "KAFKA_BATCH_SIZE",          PaInt,     PaOpt,  100,              1,      10000,            KAFKA_BATCH_SIZE_DESC        },
   { "-kafkaBatchLingerMs",    &kafkaBatchLingerMs,      "KAFKA_BATCH_LINGER_MS",     PaInt,     PaOpt,  50,               1,      5000,             KAFKA_LINGER_DESC            },
   { "-kafkaConsumerThreads",  &kafkaConsumerThreads,    "KAFKA_CONSUMER_THREADS",    PaInt,     PaOpt,  2,                1,      32,               KAFKA_THREADS_DESC           },

src/lib/orionld/common/orionldState.h

@@ -499,7 +499,7 @@
   // TRoE
   //
   bool                    noDbUpdate;            // If nothing changed in DB - troe is not invoked
-  bool                    troeError;              // Unused - TODO: remove
+  bool                    troeError;              // Set by pgCommands() when a TRoE (Postgres) write fails
   bool                    noTroe;                 // Skip TRoE for this request (e.g. local insert failed in a 207 distOp)
   KjNode*                 duplicateArray;
   KjNode*                 troeIgnoreV[20];
@@ -645,6 +645,7 @@
 extern bool              troe;                     // From orionld.cpp
 extern char              troeHost[256];            // From orionld.cpp
 extern int               troePort;                 // From orionld.cpp
+extern char              kafkaAckTopic[256];       // From orionld.cpp - Kafka ACK/NACK feedback topic (empty = disabled)
 extern char              troeUser[256];            // From orionld.cpp
 extern char              troePwd[256];             // From orionld.cpp
 extern char              troeSslMode[64];          // From orionld.cpp

src/lib/orionld/kafka/CMakeLists.txt

@@ -25,6 +25,7 @@ SET (SOURCES
     kafkaConsumerLoop.cpp
     kafkaBatchProcess.cpp
     kafkaMessageParse.cpp
+    kafkaAckProducer.cpp
     kafkaRelease.cpp
 )
 

src/lib/orionld/kafka/kafkaAckProducer.cpp

@@ -0,0 +1,244 @@
+/*
+*
+* Copyright 2026 FIWARE Foundation e.V.
+*
+* This file is part of Orion-LD Context Broker.
+*
+* Orion-LD Context Broker is free software: you can redistribute it and/or
+* modify it under the terms of the GNU Affero General Public License as
+* published by the Free Software Foundation, either version 3 of the
+* License, or (at your option) any later version.
+*
+* Orion-LD Context Broker is distributed in the hope that it will be useful,
+* but WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Affero General Public License
+* along with Orion-LD Context Broker. If not, see http://www.gnu.org/licenses/.
+*
+* For those usages not covered by this license please contact with
+* orionld at fiware dot org
+*
+* Author: Carsten Frey
+*/
+#include <librdkafka/rdkafka.h>                                // rd_kafka_*
+#include <stdlib.h>                                            // malloc, free
+#include <string.h>                                            // strlen
+#include <stdio.h>                                             // snprintf
+
+extern "C"
+{
+#include "ktrace/kTrace.h"                                     // KT_*
+}
+
+#include "orionld/common/traceLevels.h"                        // KtKafka
+#include "orionld/common/orionldState.h"                       // kafkaAckTopic (declared here, defined in orionld.cpp)
+#include "orionld/kafka/kafkaAckProducer.h"                    // Own interface
+
+
+
+// -----------------------------------------------------------------------------
+//
+// External references - kafkaBrokerList lives in orionld.cpp (not in orionldState.h)
+//
+extern char kafkaBrokerList[];
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckProducerHandle - the ACK/NACK producer (separate from the consumer handle)
+//
+static rd_kafka_t* kafkaAckProducerHandle = NULL;
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckProducerInit -
+//
+bool kafkaAckProducerInit(void)
+{
+  if (kafkaAckTopic[0] == 0)
+  {
+    KT_I("Kafka ACK/NACK feedback disabled (no -kafkaAckTopic)");
+    return true;   // feature off - not an error
+  }
+
+  char              errstr[512];
+  rd_kafka_conf_t*  conf = rd_kafka_conf_new();
+
+  if (conf == NULL)
+  {
+    KT_E("kafkaAckProducerInit: failed to create producer configuration");
+    return false;
+  }
+
+  if (rd_kafka_conf_set(conf, "bootstrap.servers", kafkaBrokerList, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
+  {
+    KT_E("kafkaAckProducerInit: failed to set bootstrap.servers: %s", errstr);
+    rd_kafka_conf_destroy(conf);
+    return false;
+  }
+
+  // The feedback record must itself be durable - idempotent producer (forces acks=all).
+  rd_kafka_conf_set(conf, "enable.idempotence", "true", errstr, sizeof(errstr));
+
+  kafkaAckProducerHandle = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
+  if (kafkaAckProducerHandle == NULL)
+  {
+    KT_E("kafkaAckProducerInit: failed to create producer: %s", errstr);  // conf freed by rd_kafka_new on failure
+    return false;
+  }
+
+  KT_I("Kafka ACK/NACK producer initialized (topic '%s')", kafkaAckTopic);
+  return true;
+}
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckProducerRelease -
+//
+void kafkaAckProducerRelease(void)
+{
+  if (kafkaAckProducerHandle == NULL)
+    return;
+
+  rd_kafka_flush(kafkaAckProducerHandle, 5000);
+  rd_kafka_destroy(kafkaAckProducerHandle);
+  kafkaAckProducerHandle = NULL;
+}
+
+
+
+// -----------------------------------------------------------------------------
+//
+// jsonEscapeAppend - minimal JSON string escaper for the short, broker-controlled error text
+//
+static void jsonEscapeAppend(char* out, int outSize, int* posP, const char* in)
+{
+  int pos = *posP;
+
+  for (const char* p = in; (*p != 0) && (pos < outSize - 2); ++p)
+  {
+    char c = *p;
+
+    if      ((c == '"') || (c == '\\')) { out[pos++] = '\\'; out[pos++] = c;   }
+    else if (c == '\n')                 { out[pos++] = '\\'; out[pos++] = 'n'; }
+    else if (c == '\r')                 { out[pos++] = '\\'; out[pos++] = 'r'; }
+    else if (c == '\t')                 { out[pos++] = '\\'; out[pos++] = 't'; }
+    else if ((unsigned char) c >= 0x20) { out[pos++] = c;                      }
+    // other control chars are dropped
+  }
+
+  *posP = pos;
+}
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckSend -
+//
+void kafkaAckSend(bool ok, int count, const char* offsetsJson, const char* batchIds, const char* error, const char* entitiesJson)
+{
+  if (kafkaAckProducerHandle == NULL)   // feature disabled
+    return;
+
+  if ((offsetsJson == NULL) || (offsetsJson[0] == 0)) offsetsJson = "[]";
+  if ((batchIds    == NULL) || (batchIds[0]    == 0)) batchIds    = "[]";
+
+  //
+  // Assemble the feedback JSON. entitiesJson is already valid JSON (kjFastRender output) and is
+  // spliced in verbatim; only 'error' (a short broker string) is escaped. bufSize is sized to hold
+  // 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;
+  size_t  errorLen    = (error != NULL) ? strlen(error) : 0;
+  int     bufSize     = (int) (256 + strlen(offsetsJson) + strlen(batchIds) + entitiesLen + errorLen * 2);
+  char*   buf         = (char*) malloc(bufSize);
+
+  if (buf == NULL)
+  {
+    KT_E("kafkaAckSend: out of memory (%d bytes)", bufSize);
+    return;
+  }
+
+  int pos = snprintf(buf, bufSize, "{\"status\":\"%s\",\"count\":%d,\"offsets\":%s,\"batchIds\":%s",
+                     ok ? "ack" : "nack", count, offsetsJson, batchIds);
+
+  if ((error != NULL) && (error[0] != 0))
+  {
+    pos += snprintf(&buf[pos], bufSize - pos, ",\"error\":\"");
+    jsonEscapeAppend(buf, bufSize, &pos, error);
+    pos += snprintf(&buf[pos], bufSize - pos, "\"");
+  }
+
+  if ((entitiesJson != NULL) && (entitiesJson[0] != 0))
+    pos += snprintf(&buf[pos], bufSize - pos, ",\"entities\":%s", entitiesJson);
+
+  pos += snprintf(&buf[pos], bufSize - pos, "}");
+
+  //
+  // RD_KAFKA_MSG_F_COPY: librdkafka copies the payload, so we can free 'buf' right after the call
+  // (without the flag, producev references the buffer until the delivery report).
+  //
+  rd_kafka_resp_err_t err = rd_kafka_producev(kafkaAckProducerHandle,
+                                              RD_KAFKA_V_TOPIC(kafkaAckTopic),
+                                              RD_KAFKA_V_VALUE(buf, strlen(buf)),
+                                              RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
+                                              RD_KAFKA_V_END);
+  if (err != RD_KAFKA_RESP_ERR_NO_ERROR)
+    KT_W("kafkaAckSend: produce to '%s' failed: %s", kafkaAckTopic, rd_kafka_err2str(err));
+  else
+    KT_T(KtKafka, "Kafka %s sent (%d entities) to '%s'", ok ? "ACK" : "NACK", count, kafkaAckTopic);
+
+  rd_kafka_poll(kafkaAckProducerHandle, 0);   // serve delivery callbacks
+  free(buf);
+}
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckSendParseFail -
+//
+void kafkaAckSendParseFail(const char* offsetsJson, const char* batchIds, const char* rawPayload, int rawLen)
+{
+  if (kafkaAckProducerHandle == NULL)   // feature disabled
+    return;
+  if ((rawPayload == NULL) || (rawLen <= 0))
+    return;
+
+  // Build ["<escaped raw>"] as a valid JSON array so it can ride in the 'entities' field verbatim.
+  int   cap = rawLen * 2 + 8;
+  char* raw = (char*) malloc(rawLen + 1);
+  char* arr = (char*) malloc(cap);
+
+  if ((raw == NULL) || (arr == NULL))
+  {
+    free(raw);
+    free(arr);
+    return;
+  }
+
+  memcpy(raw, rawPayload, rawLen);
+  raw[rawLen] = 0;
+
+  int pos = 0;
+  arr[pos++] = '[';
+  arr[pos++] = '"';
+  jsonEscapeAppend(arr, cap, &pos, raw);
+  arr[pos++] = '"';
+  arr[pos++] = ']';
+  arr[pos]   = 0;
+
+  kafkaAckSend(false, 1, offsetsJson, batchIds, "Kafka message parse failed", arr);
+
+  free(raw);
+  free(arr);
+}

src/lib/orionld/kafka/kafkaAckProducer.h

@@ -0,0 +1,85 @@
+#ifndef SRC_LIB_ORIONLD_KAFKA_KAFKAACKPRODUCER_H_
+#define SRC_LIB_ORIONLD_KAFKA_KAFKAACKPRODUCER_H_
+
+/*
+*
+* Copyright 2026 FIWARE Foundation e.V.
+*
+* This file is part of Orion-LD Context Broker.
+*
+* Orion-LD Context Broker is free software: you can redistribute it and/or
+* modify it under the terms of the GNU Affero General Public License as
+* published by the Free Software Foundation, either version 3 of the
+* License, or (at your option) any later version.
+*
+* Orion-LD Context Broker is distributed in the hope that it will be useful,
+* but WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Affero General Public License
+* along with Orion-LD Context Broker. If not, see http://www.gnu.org/licenses/.
+*
+* For those usages not covered by this license please contact with
+* orionld at fiware dot org
+*
+* Author: Carsten Frey
+*/
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckProducerInit -
+//
+// Creates the rd_kafka producer used to publish ACK/NACK feedback for KAFKA-INGESTED
+// batches. A no-op returning true when 'kafkaAckTopic' is empty (feature disabled).
+// Only ever used from the Kafka consumer path - REST ingest has its synchronous HTTP
+// response as its acknowledgement and must NOT emit to Kafka.
+//
+extern bool kafkaAckProducerInit(void);
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckProducerRelease -
+//
+extern void kafkaAckProducerRelease(void);
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckSend -
+//
+// Publishes one feedback record to 'kafkaAckTopic'. No-op if the producer is disabled.
+//
+//   ok            true  -> ACK  (batch durable in MongoDB AND TRoE)
+//                 false -> NACK (a stage failed / poison / parse error)
+//   count         number of entities the record refers to
+//   offsetsJson   rendered JSON array of {"partition":P,"offset":O} (may be NULL/"" -> "[]")
+//   batchIds      rendered JSON array of the x-batch-id headers seen (may be NULL/"" -> "[]")
+//   error         NACK only: short reason string (NULL for an ACK)
+//   entitiesJson  NACK only: the failed payload as a JSON string so it is not lost
+//                 (NULL for an ACK - a success needs no payload echo, the data is durable)
+//
+extern void kafkaAckSend(bool         ok,
+                         int          count,
+                         const char*  offsetsJson,
+                         const char*  batchIds,
+                         const char*  error,
+                         const char*  entitiesJson);
+
+
+
+// -----------------------------------------------------------------------------
+//
+// kafkaAckSendParseFail -
+//
+// NACK for a Kafka message that could not even be parsed (so there is no entity tree). The raw
+// message bytes are escaped and carried as the payload so the poison message is not lost.
+//
+extern void kafkaAckSendParseFail(const char* offsetsJson, const char* batchIds, const char* rawPayload, int rawLen);
+
+#endif  // SRC_LIB_ORIONLD_KAFKA_KAFKAACKPRODUCER_H_