Memory leak: TMQ consumer getData() never frees the native buffer from TMQGetRaw

[csuftt]

Environment

• TDengine server: 3.3.6.13
• driver-go: v3.8.2
• Go: 1.21
• OS: Linux (debian bookworm)

Problem

The TMQ data consumer leaks C-allocated memory on every data message, causing unbounded RSS growth proportional to write throughput, eventually triggering OOM.

Under ~500 QPS write load (continuous writes to a consumed data topic), the consumer process RSS grows ~470 bytes/row (~9 MB/min), leading to OOM.

Root cause

Consumer.getData() in af/tmq/consumer.go calls wrapper.TMQGetRaw(message) which internally mallocs a C buffer (tmq_get_raw), but never calls the paired wrapper.TMQFreeRaw(raw). All four return paths miss the free:

func (c *Consumer) getData(message unsafe.Pointer) ([]*tmq.Data, error) {
    errCode, raw := wrapper.TMQGetRaw(message)   // C buffer malloc'd here
    if errCode != taosError.SUCCESS {
        return nil, err                            // raw not freed
    }
    _, _, rawPtr := wrapper.ParseRawMeta(raw)
    blockInfos, err := c.dataParser.Parse(rawPtr)
    if err != nil {
        return nil, err                            // not freed
    }
    for i := 0; i < len(blockInfos); i++ {
        if err != nil { return nil, err }          // not freed
    }
    return tmqData, nil                            // not freed
}

Note: defer wrapper.TaosFreeResult(message) in Poll() only frees TAOS_RES; it does not free the separately malloc'd raw buffer (different allocation). The very existence of wrapper.TMQFreeRaw proves the pairing is required. Supporting evidence:

1. wrapper.TMQFreeRaw exists and calls tmq_free_raw (wrapper/tmq.go:206) — it must be paired with TMQGetRaw.
2. driver-go's own test pairs them (wrapper/tmq_test.go:811-818):

   errCode, rawMeta := TMQGetRaw(message)
   ...
   TMQFreeRaw(rawMeta)

3. The sibling getMeta() correctly pairs TMQGetJsonMeta/TMQFreeJsonMeta — only getData misses.

Because the C malloc'd buffer is invisible to Go GC, it accumulates until OOM.

Steps to reproduce

1. Create a data topic on a TDengine database (CREATE TOPIC ... AS SELECT * FROM <stable>).
2. Run a TMQ consumer polling that topic, processing *tmqcommon.DataMessage on each message.
3. Continuously write to the supertable at ~500 QPS.
4. Observe consumer RSS — it grows linearly (~9 MB/min at ~500 QPS) until OOM.

Fix

Add defer wrapper.TMQFreeRaw(raw) immediately after successful TMQGetRaw (mirrors getMeta's TMQFreeJsonMeta pairing):

errCode, raw := wrapper.TMQGetRaw(message)
if errCode != taosError.SUCCESS {
    return nil, err
}
defer wrapper.TMQFreeRaw(raw)   // <-- add this
_, _, rawPtr := wrapper.ParseRawMeta(raw)
...

Verification

Under the same ~500 QPS load, a TMQ consumer's RSS growth dropped from ~9 MB/min to ~0.25 MB/min (36x), stable with no OOM over 2h.

Fork with the fix: https://github.com/csuftt/driver-go/tree/v3.8.2-freeraw.1 (branch fix/freeraw, tag v3.8.2-freeraw.1)