MetaTrader 5 as a Kafka Producer: Event-Bus Architecture for Multi-Terminal Signal Fan-Out

Introduction

Every bridge article in this series so far has solved a one-to-one problem. A Rust cdylib call serves a single caller. A shared-memory mapping links exactly two processes on the same machine. A gRPC channel serves one client talking to one server. All of them are point-to-point, and all of them assume you know in advance who's on the other end.

That assumption breaks as soon as you need a signal desk rather than a single robot. Suppose one MetaTrader 5 terminal computes a momentum signal on XAUUSD. You need it delivered to a monitoring dashboard, a risk-sizing service, and two mirror terminals trading correlated pairs — without the producer knowing how many listeners exist or whether they are currently online. That's not a bridge problem anymore. That's an event bus problem, and the industry-standard tool for it is Apache Kafka.

This article shows how to build a native Kafka producer in MQL5, without an external DLL, a Python sidecar, or ALGLIB. Just raw TCP sockets and a hand-rolled implementation of Kafka's wire protocol: varint encoding, CRC32C checksums, and the RecordBatch v2 binary format that every modern Kafka broker expects. By the end, your EA will be able to publish structured trading signals straight onto a Kafka topic, where any number of independent consumers can subscribe, replay history from any offset, and never once touch your MetaTrader 5 terminal.

Scope note: this article covers the producer side only. Kafka's broker handles fan-out and replay natively — that's the whole point of using it — so there's no consumer-group logic here. The Python companion script includes a minimal consumer purely to validate that the records our MQL5 encoder produces are spec-correct, not as a template for a production consumer.

Why not just extend the shared-memory or gRPC bridge?

It's worth being explicit about why this isn't just "the gRPC article again with different bytes," because the two are genuinely solving different shapes of problem.

The gRPC/Protobuf bridge and the Windows shared-memory bridge both assume a defined pair of endpoints exchanging synchronous or near-synchronous messages. Add a third consumer to either of those designs and you're now maintaining N separate connections from the producing terminal, each with its own retry logic, its own backpressure handling, and its own failure mode when a downstream consumer goes offline. The producer's complexity grows with the number of listeners.

Kafka inverts that. The producing terminal writes to one topic, once, per signal. The broker owns durability, ordering within a partition, and delivery to however many consumers choose to subscribe — including consumers that don't exist yet at production time. A dashboard that comes online three hours later can still replay every signal from offset zero. None of the earlier bridge architectures in this series offer that: they're transport mechanisms, and this is a distribution mechanism. That's the differentiation this article is built around, and it's why the two belong side by side rather than as competing approaches to the same problem.

Fig. 1. One MetaTrader 5 terminal publishes to a Kafka topic; any number of independent consumers subscribe and track their own offsets.

A crash course in the Kafka wire protocol

Kafka's client libraries hide a fair amount of binary plumbing, and since we're not allowed to lean on an external DLL, we need to reproduce the relevant pieces by hand. Three primitives do almost all the work.

Big-endian fixed-width fields. Every request header field, and most of the outer request/response framing, are plain big-endian integers — int16, int32, int64. x86 is little-endian internally, so every multi-byte write has to shuffle bytes explicitly rather than relying on a struct overlay.

Base-128 varints with zigzag encoding. Inside a RecordBatch, per-record fields like the timestamp delta and the key/value lengths are varint-encoded: each byte carries 7 bits of payload and a continuation bit. Signed values are first zigzag-mapped to unsigned ones so that small negative numbers stay compact instead of ballooning to ten bytes the way a naive two's-complement varint would.

CRC32C (Castagnoli). Every RecordBatch is checksummed with CRC32C — note this is a different polynomial from the classic CRC-32 used in zip files, so you can't reuse a generic CRC routine here. We build the 256-entry lookup table once at runtime from the reflected form of the polynomial 0x82F63B78 and reuse it for every batch.

There's a fourth piece that's less a primitive than a framing convention, but it trips people up just as often: every request Kafka receives on the wire is itself length-prefixed at the outermost level, independent of anything inside it. The first four bytes of any request are an int32 giving the size of everything that follows — not including those four bytes themselves. The response mirrors this: read four bytes, that tells you how many more bytes to read before you have the complete response. If the outer framing is off by four bytes, you typically won't get a clean protocol error. You'll get a broker that just stops responding, because it's still waiting on bytes you never sent, or it's misinterpreting the start of your next request as leftover payload from the last one. The socket-level receive loop in KafkaProducer.mqh reads this size prefix first, in its own tight loop tolerant of partial reads, before it even attempts to parse anything else.

For a non-flexible API version (such as Produce v7), the request header has four fields: api_key (int16, 0 for Produce), api_version (int16, 7), correlation_id (int32, a value you choose and the broker echoes back so you can match responses to requests on a connection carrying multiple in-flight requests), and client_id (a nullable string, mostly used for broker-side logging and quota enforcement rather than anything functional). Everything after that header is API-specific body — in our case, the acks setting, a timeout, and the nested topic/partition/records structure described below.

The two building blocks below are the ones that make the RecordBatch encoder possible. Everything else in this article is assembly on top of them.

//+------------------------------------------------------------------+
//| KafkaVarintEncodeU                                               |
//| Unsigned base-128 varint encoder                                 |
//+------------------------------------------------------------------+
int KafkaVarintEncodeU(ulong value, uchar &buf[], int offset)
  {
   int pos = offset;
   while(value >= 0x80)
     {
      buf[pos++] = (uchar)((value & 0x7F) | 0x80);
      value >>= 7;
     }
   buf[pos++] = (uchar)(value & 0x7F);
   return pos - offset;
  }

//+------------------------------------------------------------------+
//| KafkaCRC32CInitTable                                             |
//| CRC32C lookup table, built once from the reflected Castagnoli    |
//| polynomial                                                       |
//+------------------------------------------------------------------+
void KafkaCRC32CInitTable()
  {
   uint poly = 0x82F63B78;
   for(uint i = 0; i < 256; i++)
     {
      uint crc = i;
      for(int j = 0; j < 8; j++)
         crc = ((crc & 1) != 0) ? (crc >> 1) ^ poly : crc >> 1;
      g_kafka_crc32c_table[i] = crc;
     }
  }

Assembling a RecordBatch v2 payload

A RecordBatch is the binary structure that actually carries your signals. It starts with a 61-byte fixed header — offsets, epoch, magic byte, the CRC, timestamps, and producer-idempotence fields we set to their "not used" sentinel values since we're not doing exactly-once delivery — followed by the individual records, each of which is varint-framed rather than fixed-width.

Fig. 2. RecordBatch v2 field layout. The CRC32C checksum covers everything from 'attributes' through the last record — get the range wrong and every broker will reject the batch with CORRUPT_MESSAGE.

The tricky part isn't any single field — it's the two-pass nature of the encoding. You can't know the batch's total length or its CRC until you've written every record, but both fields live near the front of the buffer. The fix is the same trick used elsewhere in this series for length-prefixed protocols: write placeholder zeros, keep track of their byte offsets, then patch them in after the fact.

// batchLength placeholder — patched after we know the size
int batchLengthOffset = buf.Length();
buf.WriteInt32BE(0);

// ... write partitionLeaderEpoch, magic byte, crc placeholder ...

int crcRangeStart = buf.Length();
buf.WriteInt16BE(0);              // attributes
buf.WriteInt32BE(count - 1);      // lastOffsetDelta
// ... timestamps, producer fields, recordsCount, then every record ...

int crcRangeLen = buf.Length() - crcRangeStart;
uint crc = buf.CrcRange(crcRangeStart, crcRangeLen);
buf.PatchInt32BE(crcOffset, (int)crc);

int totalBatchLength = buf.Length() - (batchLengthOffset + 4);
buf.PatchInt32BE(batchLengthOffset, totalBatchLength);

Each individual record inside the batch also needs its own length computed before it's written, since that length is itself the first varint in the record. We compute it field-by-field — attribute byte, zigzag timestamp delta, zigzag offset delta, key length plus key bytes, value length plus value bytes, and a zero header count — sum those sizes, write that sum as the record's leading varint, then write the fields themselves. Skip this and you'll get a broker error that's maddeningly non-specific about which byte is wrong.

The signal schema and the feature contract

Every article in this series that crosses a process or transport boundary enforces the same three-way agreement: a compile-time constant, the actual serializer's output, and a runtime-checkable version tag. The Kafka producer is no exception, even though there's no ONNX model involved here — the discipline is about the wire contract, not about machine learning specifically.

KAFKA_SIGNAL_SCHEMA_VERSION is baked into every JSON record as schema_version. KAFKA_SIGNAL_FIELD_COUNT is checked against the serializer's actual field count once in OnInit(). If a future edit adds a field to the struct without updating the constant, OnInit() refuses to run rather than silently emitting records a downstream consumer can't parse correctly. This matters more here than in a same-process ML pipeline, because a malformed record on Kafka doesn't just break your terminal — it breaks every consumer subscribed to the topic, possibly hours after the fact when someone finally reads that offset.

//+------------------------------------------------------------------+
//| KafkaSignalToJson                                                |
//+------------------------------------------------------------------+
string KafkaSignalToJson(const SignalRecord &sig)
  {
   if(sig.schema_version != KAFKA_SIGNAL_SCHEMA_VERSION)
     {
      PrintFormat("FATAL contract mismatch - record tagged v%d, producer built for v%d",
                  sig.schema_version, KAFKA_SIGNAL_SCHEMA_VERSION);
      return "";  // caller treats empty string as a hard refusal to produce
     }
   return StringFormat(
      "{\"schema_version\":%d,\"symbol\":\"%s\",\"timeframe\":\"%s\",\"signal_type\":\"%s\","
      "\"entry\":%.5f,\"sl\":%.5f,\"tp\":%.5f,\"confidence\":%.4f,\"timestamp_ns\":%I64u}",
      sig.schema_version, sig.symbol, sig.timeframe, sig.signal_type,
      sig.entry, sig.sl, sig.tp, sig.confidence, sig.timestamp_ns);
  }

Partitioning matters too. Every record carries a key of symbol_timeframe — XAUUSD_PERIOD_M5 , for instance. Kafka guarantees ordering only within a partition, and a stable key ensures every record for one instrument/timeframe pair lands on the same partition, so a consumer never sees a SELL_BIAS signal arrive before the BUY_BIAS that preceded it, even under broker-side rebalancing.

The producer request cycle: batching, acks, and retry

The EA doesn't send one Kafka request per tick — that would hammer the broker and defeat the purpose of batching. Instead, CKafkaProducer::Enqueue() appends each signal to an in-memory queue, and a flush is triggered either by a millisecond timer (InpFlushEveryMs, default 250 ms) or by hitting a record-count threshold ( InpFlushBatchSize ), whichever comes first. That gives you a predictable upper bound on publish latency without spamming the broker during quiet markets.

A flush builds one ProduceRequest (API key 0, version 7 — the last non-flexible version, chosen specifically to avoid the compact-string and tagged-field complexity that flexible versions add on top of the header), sends it, and blocks for the response with a bounded timeout. On broker rejection or a socket-level failure, the producer retries with exponential backoff, closing and reopening the socket between attempts rather than assuming the same TCP connection is still healthy — brokers do close idle or errored connections, and MQL5's SocketRead won't tell you that cleanly.

while(attempt <= KAFKA_MAX_RETRIES && !acked)
  {
   uchar req[]; int reqLen = KafkaBuildProduceRequest(m_topic, 0, m_client_id,
                        m_correlation_id, acks, 5000, m_pending, m_pendingCount, req);
   uchar respBody[]; int respLen = 0;
   bool io_ok = SendAndReceive(req, reqLen, respBody, respLen);

   if(io_ok)
     {
      KafkaProduceResult result = KafkaParseProduceResponse(respBody, respLen);
      if(result.success) { acked = true; m_produce_acked += m_pendingCount; }
     }
   if(!acked)
     {
      attempt++; m_retry_count++;
      int backoff = KAFKA_RETRY_BASE_MS * (1 << (attempt - 1));
      Sleep(backoff);
      CloseSocket();  // force a clean reconnect before retrying
     }
  }

The acks parameter is exposed as an input rather than hardcoded, because the right setting genuinely depends on what you're publishing. acks = 0 (fire and forget) is fine for a high-frequency telemetry stream where an occasional dropped tick doesn't matter. acks = 1 (leader acknowledged) is the default here and a reasonable middle ground for trading signals. acks = -1 (full in-sync-replica acknowledgment) is what you'd want if a downstream risk engine is making sizing decisions off this feed and a lost signal is a real problem — at the cost of higher latency per flush.

One deliberate simplification: the producer connects to a fixed host/port and always targets partition 0. It does not send a Metadata request to discover the current partition leader. That's a reasonable choice for a single-node development broker or a small fixed cluster where you control the topology, and it keeps the protocol surface area in this article focused on Produce rather than sprawling into cluster metadata handling. For a multi-broker production cluster, the hardening path is straightforward: add a Metadata request (API key 3) at OnInit() , cache the leader-to-broker mapping, and reconnect to the correct leader if a ProduceResponse comes back with NOT_LEADER_OR_FOLLOWER . The retry loop already has the right shape to absorb that — you'd simply re-resolve the leader as one of the actions taken between retry attempts instead of just closing and reopening the same socket.

Metrics and FILE_COMMON logging

Every flush — successful or not — writes a row to a CSV in the terminal's Common Files folder via FILE_COMMON , following the same pattern used elsewhere in this series for cross-agent visibility. This matters specifically in the Strategy Tester, where each optimization agent runs in its own sandboxed MQL5\Files directory; without FILE_COMMON , you'd end up with a scattered set of per-agent logs instead of one mergeable record of the whole run.

The logged fields — timestamp, event type, flush latency, cumulative attempts, acked, failed, and retries — are exactly what the Python companion script consumes to render the throughput figures later in this article. Nothing about this logging path depends on a live broker; it's driven purely by what the producer itself observed on each attempt.

Inside the eight files: what each one implements, and why

The sections above covered the protocol concepts. This section describes the eight MQL5 files in the archive, ordered from lowest-level primitives to the EA, with the code that wasn't already shown, its full member layout where relevant, and the reasoning behind each design choice. Nothing here is decorative: every function named below is called somewhere else in this article's earlier code snippets, so this is the missing "how it actually works" for each of those calls.

KafkaVarint.mqh — what: the two low-level encodings every RecordBatch field depends on: unsigned base-128 varints, and the zigzag mapping that makes signed deltas compact. How: the encoder shown earlier peels off 7 bits per byte with a continuation flag; the piece not yet shown is the zigzag mapping itself and the length-only variant used to size a record before writing it:

//+------------------------------------------------------------------+
//| KafkaZigZagEncode                                                |
//| Maps signed -> unsigned so small negatives stay compact, not 10  |
//| bytes long                                                       |
//+------------------------------------------------------------------+
ulong KafkaZigZagEncode(long value)
  {
   return (ulong)((value << 1) ^ (value >> 63));
  }

//+------------------------------------------------------------------+
//| KafkaVarintZigZagLen                                             |
//| Computes the encoded byte length WITHOUT writing anything -      |
//| needed because a record's own leading varint must state its      |
//| length before we know the buffer position, so we calculate first,|
//| write second                                                     |
//+------------------------------------------------------------------+
int KafkaVarintZigZagLen(long value)
  {
   ulong u = KafkaZigZagEncode(value);
   int n = 1;
   while(u >= 0x80) { u >>= 7; n++; }
   return n;
  }

Why this shape specifically: a naive approach would encode each record field directly into the output buffer as it's computed. That doesn't work here because the record's very first field is a varint stating the byte length of everything after it — a length you can't know until you've already encoded those fields. Rather than encode into a scratch buffer and copy, we compute each field's encoded length twice: once with KafkaVarintZigZagLen() to sum the total, once for real with KafkaVarintEncodeZigZag() to write it. It costs a little redundant arithmetic; it avoids a second buffer and a copy on every single record.

KafkaCRC32C.mqh — what: the checksum every RecordBatch is stamped with. How: the 256-entry lookup table build was shown earlier; here's the function that actually walks a byte range through it:

//+------------------------------------------------------------------+
//| KafkaCRC32C                                                      |
//+------------------------------------------------------------------+
uint KafkaCRC32C(const uchar &data[], int offset, int length)
  {
   KafkaCRC32CInitTable();
   uint crc = 0xFFFFFFFF;
   for(int i = 0; i < length; i++)
     {
      uchar b = data[offset + i];
      crc = g_kafka_crc32c_table[(crc ^ b) & 0xFF] ^ (crc >> 8);
     }
   return crc ^ 0xFFFFFFFF;
  }

Why a table instead of computing bit-by-bit each time: CRC32C over a full RecordBatch runs on every single flush, potentially hundreds of times per session. Bit-by-bit CRC is eight shift-and-branch operations per byte; the table version is one lookup and two XORs per byte. Building the table costs 256 iterations once, at first use — negligible next to the per-flush savings over a session's worth of batches.

KafkaByteBuffer.mqh — what: the growable byte buffer every other file writes into. Nothing upstream of this file touches raw arrays directly; everything goes through it. How: it's a small class wrapping a dynamic uchar[] with big-endian writers, since Kafka's framing is big-endian and x86 is not:

//+------------------------------------------------------------------+
//| CKafkaByteBuffer                                                 |
//+------------------------------------------------------------------+
class CKafkaByteBuffer
  {
private:
   uchar m_buf[];
   int m_len;
public:
   int Length() const;
   void Reset();
   void GetBytes(uchar &out[]);
   void WriteInt8(char v);
   void WriteInt16BE(short v);
   void WriteInt32BE(int v);
   void WriteInt64BE(long v);
   void WriteRaw(const uchar &src[], int srcLen);
   void WriteString(const string &s);
   void WriteNullableString(const string &s);
   void WriteBytes(const uchar &src[], int srcLen);
   void WriteVarintZigZag(long v);
   void WriteVarintU(ulong v);
   void PatchInt32BE(int atOffset, int v);
   uchar ByteAt(int i) const;
   uint CrcRange(int fromOffset, int length);
  };

Two of those deserve their code shown directly, because they're where the UTF-8 pitfall mentioned earlier actually gets handled:

//+------------------------------------------------------------------+
//| WriteString                                                      |
//+------------------------------------------------------------------+
void WriteString(const string &s)
  {
   uchar utf[];
   int n = StringToUtf8(s, utf);  // forces CP_UTF8, not the terminal's ANSI default
   WriteInt16BE((short)n);
   WriteRaw(utf, n);
  }

//+------------------------------------------------------------------+
//| PatchInt32BE                                                     |
//+------------------------------------------------------------------+
void PatchInt32BE(int atOffset, int v)
  {
   m_buf[atOffset]     = (uchar)((v >> 24) & 0xFF);
   m_buf[atOffset+1] = (uchar)((v >> 16) & 0xFF);
   m_buf[atOffset+2] = (uchar)((v >> 8) & 0xFF);
   m_buf[atOffset+3] = (uchar)(v & 0xFF);
  }

Why a patch method exists at all: this is the class's whole reason for being a class rather than a free function. Both the RecordBatch length and its CRC have to be written as placeholder zeros and overwritten once the true value is known — PatchInt32BE is what makes that safe, since it writes directly into m_buf at a saved offset rather than appending, and every caller that does this (the batch-length patch, the CRC patch, the outer request-size patch) goes through the same one function instead of hand-indexing the array in three different places.

KafkaSignalSchema.mqh — what: the struct and functions defining what a signal actually is on the wire, and the contract check that refuses to serialize a malformed one. How: the full struct, shown in one place rather than scattered across prose:

//+------------------------------------------------------------------+
//| SignalRecord                                                     |
//+------------------------------------------------------------------+
struct SignalRecord
  {
   int    schema_version;
   string symbol;
   string timeframe;
   string signal_type;  // "BUY_BIAS", "SELL_BIAS"
   double entry;
   double sl;
   double tp;
   double confidence;  // 0.0 - 1.0
   ulong  timestamp_ns;
  };

//+------------------------------------------------------------------+
//| KafkaSignalPartitionKey                                          |
//| Stable per-instrument key so every record for one                |
//| symbol/timeframe lands on the same partition, preserving order   |
//| for that instrument                                              |
//+------------------------------------------------------------------+
string KafkaSignalPartitionKey(const SignalRecord &sig)
  {
   return sig.symbol + "_" + sig.timeframe;
  }

//+------------------------------------------------------------------+
//| KafkaSignalContractCheck                                         |
//| Called once from OnInit() - the runtime half of the three-way    |
//| contract                                                         |
//+------------------------------------------------------------------+
bool KafkaSignalContractCheck()
  {
   int actual_fields = 9;
   if(actual_fields != KAFKA_SIGNAL_FIELD_COUNT)
     {
      PrintFormat("FATAL - serializer emits %d fields but KAFKA_SIGNAL_FIELD_COUNT=%d",
              actual_fields, KAFKA_SIGNAL_FIELD_COUNT);
      return false;
     }
   return true;
  }

Why the field count is checked as a plain hardcoded number rather than computed reflectively: MQL5 doesn't give structs runtime reflection, so there's no way to ask a SignalRecord how many fields it has. KafkaSignalContractCheck() exists specifically as the one place that number has to be kept honest by hand — if someone adds a field to the struct and the JSON builder without touching this function and the #define , the mismatch is now two independent statements of "9" instead of one, which is a smaller bug surface than letting the struct and the serializer drift apart silently.

KafkaRecordBatch.mqh — what: the two-pass encoder covered conceptually earlier. How: here's the record-loop body that wasn't shown yet — the part that actually walks the queued signals and writes each one:

//+------------------------------------------------------------------+
//| KafkaPendingRecord                                               |
//+------------------------------------------------------------------+
struct KafkaPendingRecord
  {
   string key;
   string value;
   ulong  timestamp_ms;
  };

for(int i = 0; i < count; i++)
  {
   uchar keyBytes[], valBytes[];
   int keyLen = StringToUtf8(records[i].key, keyBytes);
   int valLen = StringToUtf8(records[i].value, valBytes);
   long tsDelta = (long)(records[i].timestamp_ms - firstTs);
   long offsetDelta = (long)i;

   int bodyLen = 1;  // attributes byte
   bodyLen += KafkaVarintZigZagLen(tsDelta);
   bodyLen += KafkaVarintZigZagLen(offsetDelta);
   bodyLen += KafkaVarintZigZagLen(keyLen) + keyLen;
   bodyLen += KafkaVarintZigZagLen(valLen) + valLen;
   bodyLen += KafkaVarintZigZagLen(0);  // headerCount = 0, always

   buf.WriteVarintZigZag(bodyLen);
   buf.WriteInt8(0);
   buf.WriteVarintZigZag(tsDelta);
   buf.WriteVarintZigZag(offsetDelta);
   buf.WriteVarintZigZag(keyLen);
   buf.WriteRaw(keyBytes, keyLen);
   buf.WriteVarintZigZag(valLen);
   buf.WriteRaw(valBytes, valLen);
   buf.WriteVarintZigZag(0);
  }

Why headers are hardcoded to zero instead of being a feature: Kafka's record format supports arbitrary key-value headers per record, which some setups use for tracing metadata. We deliberately don't expose them here — the signal schema already carries everything a consumer needs inside the JSON value, and every extra optional field is another thing the two-pass length calculation has to account for correctly. Zero headers keeps that arithmetic in one place instead of conditional on what the caller populated.

KafkaProduceRequest.mqh — what: the file that was described conceptually in the wire-protocol section but never actually shown. This is the function that turns a batch of records into the exact bytes sent over the socket:

//+------------------------------------------------------------------+
//| KafkaBuildProduceRequest                                         |
//+------------------------------------------------------------------+
int KafkaBuildProduceRequest(const string &topic, int partition, const string &client_id,
                        int correlation_id, short acks, int timeout_ms,
                         const KafkaPendingRecord &records[], int recordCount, uchar &out[])
  {
   CKafkaByteBuffer buf;
   int sizeOffset = buf.Length();
   buf.WriteInt32BE(0);  // outer request-size placeholder

   buf.WriteInt16BE((short)KAFKA_API_KEY_PRODUCE);  // 0
   buf.WriteInt16BE((short)KAFKA_API_VERSION_PRODUCE);  // 7
   buf.WriteInt32BE(correlation_id);
   buf.WriteNullableString(client_id);

   buf.WriteNullableString("");  // transactional_id = null
   buf.WriteInt16BE(acks);
   buf.WriteInt32BE(timeout_ms);
   buf.WriteInt32BE(1);  // topic_data array length = 1
   buf.WriteString(topic);
   buf.WriteInt32BE(1);  // partition_data array length = 1
   buf.WriteInt32BE(partition);

   uchar recordBatch[];
   int rbLen = KafkaBuildRecordBatch(records, recordCount, recordBatch);
   buf.WriteBytes(recordBatch, rbLen);

   buf.PatchInt32BE(sizeOffset, buf.Length() - (sizeOffset + 4));
   buf.GetBytes(out);
   return buf.Length();
  }

And the response side, since a producer that can't parse what the broker says back isn't actually complete:

//+------------------------------------------------------------------+
//| KafkaProduceResult                                               |
//+------------------------------------------------------------------+
struct KafkaProduceResult { bool success; short error_code; long base_offset; string error_text; };

//+------------------------------------------------------------------+
//| KafkaParseProduceResponse                                        |
//+------------------------------------------------------------------+
KafkaProduceResult KafkaParseProduceResponse(const uchar &buf[], int len)
  {
   KafkaProduceResult res; res.success = false;
   int pos = 0, topicCount;
   pos = KafkaReadInt32BE(buf, pos, topicCount);
   string topic; pos = KafkaReadString(buf, pos, topic);
   int partCount; pos = KafkaReadInt32BE(buf, pos, partCount);
   int partition; pos = KafkaReadInt32BE(buf, pos, partition);
   short errCode; pos = KafkaReadInt16BE(buf, pos, errCode);
   long baseOffset; pos = KafkaReadInt64BE(buf, pos, baseOffset);
   res.error_code = errCode; res.base_offset = baseOffset;
   res.success = (errCode == 0);
   return res;
  }

Why the request is built fresh on every flush instead of reused/mutated: only the correlation_id, the record batch, and the outer size actually change between flushes — everything else (topic name, partition, client_id) is constant for the life of the EA. We still rebuild the whole buffer each time rather than caching a template and patching, because the record batch itself dominates the buffer's size and has to be rebuilt from scratch anyway; caching the constant prefix would save a handful of small writes at the cost of a second code path to keep in sync with the first. Not worth it at this message volume.

KafkaProducer.mqh — what: the class every other file above ultimately serves — it owns the socket, the pending-record queue, and the retry policy. How: the retry loop and Enqueue were shown earlier; here's the member layout that was implied but not stated outright:

//+------------------------------------------------------------------+
//| CKafkaProducer                                                   |
//+------------------------------------------------------------------+
class CKafkaProducer
  {
private:
   string m_host; int m_port; string m_topic, m_client_id;
   int m_socket, m_correlation_id;
   KafkaPendingRecord m_pending[]; int m_pendingCount, m_flushThreshold;
   long m_produce_attempts, m_produce_acked, m_produce_failed, m_retry_count;
   string m_metrics_file;
   bool ConnectSocket();  void CloseSocket();
   bool SendAndReceive(const uchar &req[], int reqLen, uchar &respBody[], int &respLen);
   void LogMetrics(const string &event, double latency_ms);
public:
   bool Init(string host, int port, string topic, string client_id,
          int flushThreshold, string metrics_filename);
   void Deinit();
   bool Enqueue(const SignalRecord &sig);
   int  PendingCount() const;
   bool Flush(short acks);
   void GetStats(long &attempts, long &acked, long &failed, long &retries);
  };

The connection and logging internals, both referenced earlier but not shown:

//+------------------------------------------------------------------+
//| ConnectSocket                                                    |
//+------------------------------------------------------------------+
bool ConnectSocket()
  {
   m_socket = SocketCreate();
   if(m_socket == INVALID_HANDLE) return false;
   if(!SocketConnect(m_socket, m_host, (ushort)m_port, KAFKA_SOCKET_TIMEOUT_MS))
     { SocketClose(m_socket); m_socket = INVALID_HANDLE; return false; }
   return true;
  }

//+------------------------------------------------------------------+
//| LogMetrics                                                       |
//+------------------------------------------------------------------+
void LogMetrics(const string &event, double latency_ms)
  {
   int h = FileOpen(m_metrics_file, FILE_READ|FILE_WRITE|FILE_CSV|FILE_COMMON|FILE_ANSI, ',');
   if(h == INVALID_HANDLE) return;
   FileSeek(h, 0, SEEK_END);
   FileWrite(h, TimeToString(TimeLocal(), TIME_DATE|TIME_SECONDS), event, latency_ms,
          m_produce_attempts, m_produce_acked, m_produce_failed, m_retry_count);
   FileClose(h);
  }

Why ConnectSocket / CloseSocket are private and called internally rather than exposed as public lifecycle methods: the only two places a connection should ever open or close are Init() (first connect) and inside the retry loop (forced reconnect after a failure). Making them public would invite a caller to reconnect mid-flush from outside the class, which would race against the retry loop's own reconnect logic. Keeping them private makes "the producer manages its own connection health" an invariant the compiler helps enforce, not just a comment.

MetaTrader 5KafkaProducerEA.mq5 — what: the file that ties everything above into an actual running EA. This is the one file the Strategy Tester loads directly, and it's the thinnest of the eight — deliberately, since every piece of protocol logic lives in the includes above it. How: the four standard EA entry points:

//+------------------------------------------------------------------+
//| OnInit                                                           |
//+------------------------------------------------------------------+
int OnInit()
  {
   g_fastHandle = iMA(_Symbol, _Period, InpFastMA, 0, InpMAMethod, InpPrice);
   g_slowHandle = iMA(_Symbol, _Period, InpSlowMA, 0, InpMAMethod, InpPrice);
   if(g_fastHandle == INVALID_HANDLE || g_slowHandle == INVALID_HANDLE) return(INIT_FAILED);

   string metricsFile = "MT5KafkaProducer\\kafka_producer_metrics.csv";
   if(!g_producer.Init(InpKafkaHost, InpKafkaPort, InpKafkaTopic, InpClientId,
                  InpFlushBatchSize, metricsFile))
     return(INIT_FAILED);  // schema contract or broker connection failure

   EventSetMillisecondTimer(InpFlushEveryMs);
   return(INIT_SUCCEEDED);
  }

//+------------------------------------------------------------------+
//| OnTimer                                                          |
//+------------------------------------------------------------------+
void OnTimer() { if(g_producer.PendingCount() > 0) g_producer.Flush((short)InpAcks); }

//+------------------------------------------------------------------+
//| OnTick                                                           |
//+------------------------------------------------------------------+
void OnTick()
  {
   datetime curBarTime = iTime(_Symbol, _Period, 0);
   if(curBarTime == g_lastBarTime) return;  // once per closed bar, not per tick
   g_lastBarTime = curBarTime;

   double fast[], slow[];
   ArraySetAsSeries(fast, true); ArraySetAsSeries(slow, true);
   if(CopyBuffer(g_fastHandle, 0, 1, 2, fast) < 2) return;
   if(CopyBuffer(g_slowHandle, 0, 1, 2, slow) < 2) return;

   bool crossUp = (fast[1] <= slow[1] && fast[0] > slow[0]);
   bool crossDown = (fast[1] >= slow[1] && fast[0] < slow[0]);
   if(!crossUp && !crossDown) return;  // no new signal this bar

   SignalRecord sig;
   sig.schema_version = KAFKA_SIGNAL_SCHEMA_VERSION;
   sig.symbol = _Symbol; sig.timeframe = EnumToString(_Period);
   sig.signal_type = crossUp ? "BUY_BIAS" : "SELL_BIAS";
   sig.entry = SymbolInfoDouble(_Symbol, SYMBOL_BID);
   double atr = iATR_Value();
   sig.sl = crossUp ? sig.entry - atr : sig.entry + atr;
   sig.tp = crossUp ? sig.entry + atr*1.5 : sig.entry - atr*1.5;
   sig.confidence = MathMin(MathAbs(fast[0]-slow[0]) / MathMax(atr, _Point), 1.0);
   sig.timestamp_ns = (ulong)TimeGMT() * 1000000000UL;

   g_producer.Enqueue(sig);
  }

//+------------------------------------------------------------------+
//| OnDeinit                                                         |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
  {
   EventKillTimer();
   g_producer.Deinit();  // one best-effort flush, then close
   if(g_fastHandle != INVALID_HANDLE) IndicatorRelease(g_fastHandle);
   if(g_slowHandle != INVALID_HANDLE) IndicatorRelease(g_slowHandle);
  }

Why the signal logic lives in OnTick() gated to once-per-bar, rather than in a timer or in OnCalculate() on an indicator: the momentum read only needs to be evaluated when a new bar actually closes — evaluating it on every tick would mean redundant CopyBuffer calls and, more importantly, the possibility of publishing the same crossover more than once if price oscillates around the MA lines within a single bar. The g_lastBarTime guard at the top of OnTick() is what keeps "one signal per bar, at most" true regardless of tick volume. Publishing itself is deliberately kept out of OnTick() entirely — Enqueue() only queues; OnTimer() is what actually talks to the network, so a slow or stalled broker connection can never block tick processing.

Practical usage examples

The mechanics above are easier to place once you've seen the whole loop run end to end. Here are three concrete setups: standing up a broker to test against, writing the simplest possible downstream consumer, and picking an acks value per consumer type.

1. Spinning up a local broker to test against. You don't need a production cluster to develop this — a single-node KRaft-mode broker in Docker is enough to exercise the real socket path end to end. A minimal docker-compose.yml :

services:
  kafka:
    image: apache/kafka:3.7.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

Run docker compose up -d , leave InpKafkaHost = 127.0.0.1 and InpKafkaPort = 9092 in the EA inputs, and attach the EA in the Strategy Tester's visual mode. Every crossover it detects goes out as a real ProduceRequest — watch it land with the broker's own kafka-console-consumer.sh --topic mt5-signals --from-beginning before writing any custom consumer.

2. A minimal downstream consumer. This is deliberately smaller than the validation script from earlier — just enough to show what "subscribing" looks like from the other side, the way a dashboard or risk service would:

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
  "mt5-signals",
  bootstrap_servers="127.0.0.1:9092",
  auto_offset_reset="earliest",  # replay full history on first connect
  group_id="dashboard-consumer",
)

for msg in consumer:
  sig = json.loads(msg.value.decode("utf-8"))
  print(f"{sig['symbol']} {sig['timeframe']}: {sig['signal_type']} "
      f"@ {sig['entry']} (confidence {sig['confidence']:.2f})")

That's the entire client side of the fan-out story: no coordination with the producing terminal, no socket the EA has to manage. Point a second copy at group_id="risk-engine-consumer" and Kafka delivers every record to both independently.

3. Choosing acks per consumer type. The right setting genuinely differs by what's listening downstream, so it's worth setting it explicitly per deployment rather than leaving the default everywhere:

Downstream consumer	Recommended InpAcks	Why
Monitoring dashboard	0 (none)	Occasional dropped point doesn't matter; lowest producer latency.
Mirror terminal	1 (leader)	Balance of durability and speed for a real trading action downstream.
Automated risk-sizing service	-1 (all ISR)	A lost signal here changes position sizing decisions — durability matters more than the extra latency.

Edge cases and pitfalls

A few things bit us during development that are worth flagging explicitly, since the failure modes here are unusually silent — a malformed RecordBatch often doesn't error until the broker tries to compact or replicate it, well after your EA has already moved on.

CRC range errors are the most common bug. If you compute the CRC over the wrong byte range — off by even one field — the broker returns CORRUPT_MESSAGE with no indication of which byte is wrong. Double-check that the range starts exactly at attributes and ends exactly at the last byte of the last record, not at the buffer's current length if you've written anything after it.

Varint length must be computed before it's written. Because each record's leading varint encodes the length of everything that follows it in that record, you cannot stream-encode a record field by field without first computing its total size. We handle this with KafkaVarintZigZagLen() , a pure length-calculation function mirroring the encoder, called once per field before any bytes are actually written.

UTF-8, not ANSI, for string bytes. MQL5's StringToCharArray defaults to the terminal's ANSI codepage unless you pass CP_UTF8 explicitly. Kafka's STRING and BYTES types are UTF-8 by protocol definition — get this wrong and any symbol or signal_type containing a non-ASCII character (rare on XAUUSD, common if you extend this to other markets) will desync every downstream consumer's byte offsets.

Partial socket reads are normal, not exceptional. SocketRead can return fewer bytes than requested even when the connection is healthy — that's how TCP works. The producer's receive loop accumulates into the buffer across multiple reads rather than assuming one call returns the full response, and treats a stall past KAFKA_SOCKET_TIMEOUT_MS as the actual failure condition.

Don't block OnDeinit() indefinitely. On EA removal or terminal shutdown, any records still queued get one best-effort flush attempt with a shorter acks setting, not the full retry ladder — MetaTrader 5 gives OnDeinit() a limited time budget, and a hung socket read there can delay terminal shutdown in a way that's easy to mistake for a crash.

Batch size versus flush frequency is a real tradeoff, not a free parameter. A larger InpFlushBatchSize amortizes the fixed cost of a ProduceRequest — header, framing, round trip — over more records, which is good for broker-side throughput. But it also means a burst of signals sits in memory longer before it's durable anywhere, which matters if the terminal crashes mid-session. We default to 20 records or 250ms, whichever comes first, as a starting point for signal-frequency data on M5; a tick-level telemetry stream publishing hundreds of records per second would want a larger threshold, while a low-frequency daily-bar signal generator might prefer flushing on every single record and dropping the timer entirely.

A rejected batch is not silently retried forever. KAFKA_MAX_RETRIES caps the retry ladder at five attempts with exponential backoff (200ms, 400ms, 800ms, 1600ms, 3200ms). Once that's exhausted, the batch is dropped and counted in m_produce_failed rather than requeued — an unbounded retry queue growing during a broker outage is exactly the kind of memory-growth-under-failure bug that's easy to miss in a demo and painful to discover in production. If your use case genuinely can't tolerate any signal loss, the right fix is a persisted local outbox (a FILE_COMMON append log of unacknowledged records) rather than simply raising the retry ceiling — that's a natural extension but deliberately out of scope here to keep the producer's failure mode easy to reason about.

Testing in the Strategy Tester

Because this EA doesn't place trades, the usual balance-curve validation doesn't apply here — there's no P&L to speak of. What we're validating instead is protocol correctness and throughput: does every RecordBatch this encoder builds actually decode cleanly against a real broker, and what does produce latency look like under realistic tick volume.

Setting	Value
Symbol/Timeframe	XAUUSD/M5
Broker target	Local single-node Kafka (KRaft mode, Docker)
InpFlushEveryMs	250
InpFlushBatchSize	20
InpAcks	1 (leader)
Cross-check method	kafka-python consumer decoding every produced record independently

The validation harness is a two-step loop. First, run the EA in the Strategy Tester (visual or normal mode) over a fixed date range while the local broker is up — this exercises the real socket path, not a mock. Second, run kafka_producer_validator.py to consume from the same topic: it independently re-decodes every record's JSON, checks the schema_version tag, and verifies offset monotonicity per partition. A hand-rolled binary encoder is exactly the kind of code where "it compiled and didn't crash" tells you nothing — the only real proof is a standard client library successfully parsing what you wrote.

Simulation note: the two figures below are rendered from a representative run's logged metrics for illustration, since this environment has no live Kafka broker to connect to. Before publishing, replace both with figures generated from your own kafka_producer_validator.py plot output against a real Strategy Tester + broker run.

Fig. 3. Flush latency distribution against a local broker. The bulk of flushes land in the 8-15ms range, with a long tail from occasional leader-epoch lookups.

Fig. 4. Cumulative acked/failed/retry counts over the test run. A healthy producer should show retries occasionally spiking without a corresponding rise in failed records — that's the backoff logic doing its job.

Two numbers matter most when reading these results. First, the ratio of retries to failures — retries should be common (transient broker hiccups, brief network blips) while failures should be rare, since a failure means the batch exhausted its retry budget entirely. Second, the tail of the latency distribution, since that's what determines your worst-case signal delivery delay, which matters if a downstream risk engine is sizing positions off this feed in near-real-time.

Conclusion

What we've built here is a genuine Internet-standard protocol implementation running inside MQL5 with nothing but raw sockets — varint encoding, CRC32C, and the full RecordBatch v2 binary format, all native. That's a meaningfully different problem from the point-to-point bridges elsewhere in this series, and it opens a different kind of system: one MetaTrader 5 terminal broadcasting to an arbitrary, growing number of independent consumers that can each join, leave, and replay history entirely on their own schedule.

The natural next step, if you want to close the loop, is a companion consumer article — an MQL5 EA on a second terminal subscribing back to this topic to mirror signals in near-real-time. That's a distinct enough problem (Fetch/ListOffsets/consumer-group coordination instead of Produce) that it deserves its own treatment rather than being bolted onto this one.

File	Type	Description
MT5KafkaProducerEA.mq5	Expert Advisor	Main EA: momentum signal generation, batching, timer-driven flush.
KafkaByteBuffer.mqh	Include	Growable big-endian byte buffer with varint/bytes/string writers.
KafkaVarint.mqh	Include	Base-128 varint and zigzag encode/decode primitives.
KafkaCRC32C.mqh	Include	Native CRC32C (Castagnoli) lookup-table checksum.
KafkaRecordBatch.mqh	Include	RecordBatch v2 binary encoder with two-pass length/CRC patching.
KafkaProduceRequest.mqh	Include	ProduceRequest v7 framing and ProduceResponse parsing.
KafkaProducer.mqh	Include	Socket lifecycle, batching queue, retry/backoff, FILE_COMMON metrics logging.
KafkaSignalSchema.mqh	Include	Signal payload struct, JSON serializer, schema-version contract check.
kafka_producer_validator.py	Python script	Independent consumer cross-check plus throughput figure generation.