Someone asked me a simple question about the Grafana Kafka Datasource plugin: how many messages per second can it actually handle?
I had no honest answer. I had a feeling, and feelings are not numbers. Worse, I had a vague sense that "it's probably fine," which was hiding a bug that made Protobuf decoding roughly twelve times slower than it needed to be.
This post is not really about my plugin. It's about the loop I used to find that bug: benchmark, profile, fix, prove. The plugin is just a convenient patient on the table. If you maintain any Go service that touches data on a hot path, the same four steps apply, and you can steal the commands verbatim.
First, what does this plugin actually do?
If you haven't used it, a quick orientation, because everything below happens inside one specific loop and the loop is the whole story.
The Kafka datasource is a backend Grafana plugin, which means Grafana's server runs it as a separate process (your browser never talks to Kafka directly). When you add a panel with a Kafka query, Grafana opens a Grafana Live channel and calls the plugin's RunStream. From that point, the plugin is a Kafka consumer with a WebSocket on the other end:
The interesting part is what happens per message. Every single Kafka message that arrives goes through four stages before it reaches the browser:
Nested JSON gets flattened so {"user": {"name": "x"}} becomes a user.name column. Numeric fields become time series, string fields become labels. Then each message becomes a one-row data frame and gets pushed down the Live channel.
Notice the shape of that: every cost in this pipeline is paid once per message. A wasteful millisecond in a batch job is a rounding error. A wasteful millisecond here gets multiplied by your topic's throughput. That's why the bugs below hurt as much as they did, and it's the reason this pipeline was worth profiling at all.
Why not just start optimizing?
Because you'll be wrong; I was.
If you'd asked me to guess where the plugin spent its per-message time, I'd have said "decoding bytes." That's the part that looks expensive. The profiler said otherwise: byte decoding was almost free. The cost was in compiling the schema, over and over, once for every message.
You cannot guess your way to that. So, the loop:
Four steps, four different questions:
Benchmark = how fast is it right now, in a number I can reproduce?
Profile = which line is responsible?
Fix = change only what the data justifies.
Prove = is it faster beyond measurement noise, or did I get one lucky run?
Skip step four, and you're not optimizing; you're doing performance astrology.
Step 0: Make your benchmark honest first
Before any of that, one piece of housekeeping that cost me an embarrassing amount of confusion.
The plugin calls log.DefaultLogger.Debug(...) on every message. That's fine in production at normal volumes. Inside a benchmark that runs the same function a few million times, it floods stdout and completely distorts the timing. My first "results" were mostly measuring how fast my terminal could scroll.
The fix is one line in a TestMain:
func TestMain(m *testing.M) {
log.DefaultLogger = log.NewNullLogger()
os.Exit(m.Run())
}
Both benchmark packages now do this. It's benchmark hygiene, but it doubles as a real observation about production: per-message debug logging isn't free either.
Before you trust any benchmark number, ask what else your code does on the hot path that isn't the thing you're measuring. Logging, metrics, tracing. They all show up.
Step 1: Write benchmarks that match reality
Micro-benchmarks isolate one function. I wrote them for the pieces I suspected: DecodeAvroMessage, DecodeProtobufMessage, ParseProtobufSchema.
But micro-benchmarks lie by omission. A function can be fast while the pipeline around it is slow. So I also benchmarked the full per-message path from the diagram above, across the whole format matrix the plugin supports (plaintext, lineprotocol, avro, protobuf, json_20fields, json_100fields) and through two sink modes: noop for processing only, and sendframe_json for processing plus real serialization cost.
That's BenchmarkWorkflow, and running it is ordinary Go:
go test -run '^$' -bench BenchmarkWorkflow -benchmem ./pkg/plugin/...
The -run '^$' matches no tests, so you get benchmarks only. -benchmem adds B/op and allocs/op, which matter more than you'd think: allocation count is often the real story behind a slow function.
All numbers below come from go test -bench on an Apple M5. Yours will differ. The relative improvements should hold.
Step 2: Profile, and read the profile properly
Here's where most people stop too early. go tool pprof -top gives you a function list, and a function list is not an answer. You want the line.
# Generate profiles alongside the benchmark
go test -run '^$' -bench BenchmarkWorkflow -benchmem \
-cpuprofile workflow.cpu.prof \
-memprofile workflow.mem.prof \
./pkg/plugin/...
# Function-level overview
go tool pprof -top workflow.cpu.prof
# The useful one: annotated source, line by line
go tool pprof -list 'DecodeAvroMessage' workflow.cpu.prof
# Allocation counts, not just bytes
go tool pprof -top -alloc_objects workflow.mem.prof
Wait, alloc_objects or alloc_space? alloc_space tells you how many bytes you allocated. alloc_objects tells you how many times you called the allocator. They point at different bugs. A million tiny allocations barely move alloc_space but wreck your GC pressure. Look at both.
And if you want the flame graph:
go tool pprof -http=:0 workflow.cpu.prof
-list is what cracked this open. It pointed at a single line inside the decode helper, and that line wasn't decoding anything.
Step 3: Fix
Fix #1: The Avro codec was recompiled on every message
The problem. DecodeAvroMessage called goavro.NewCodec(schema), which parses the Avro schema JSON and builds an internal codec, on every single message. The schema string never changes between messages on the same topic. We were rebuilding the same object, forever.
// before
func DecodeAvroMessage(data []byte, schema string) (interface{}, error) {
codec, err := goavro.NewCodec(schema) // recompiled every call
...
}
If that makes you wince, good. It's the kind of bug that's invisible in code review, because the line looks perfectly reasonable in isolation, and screamingly obvious in a profile.
The fix. A *goavro.Codec is immutable once built and safe for concurrent reuse, so it can simply be cached per schema string. I used a bounded LRU (default 256 entries) so a deployment with high schema churn can't grow the cache without limit:
var avroCodecCache = newLRUCache[*goavro.Codec](
cacheSizeFromEnv("KAFKA_DS_PERF_AVRO_CODEC_CACHE_MAX_ENTRIES", 256),
)
func getAvroCodec(schema string) (*goavro.Codec, error) {
if perfflags.AvroCodecCache.Disabled() {
return goavro.NewCodec(schema) // pre-fix behavior, on demand
}
if cached, ok := avroCodecCache.Get(schema); ok {
return cached, nil
}
codec, err := goavro.NewCodec(schema)
if err != nil {
return nil, err
}
avroCodecCache.Add(schema, codec)
return codec, nil
}
Results:
| Benchmark |
Before |
After |
Change |
DecodeAvroMessage (ns/op) |
5,166 ns |
341 ns |
-93.4% |
DecodeAvroMessage (B/op) |
8,260 B |
760 B |
-90.8% |
DecodeAvroMessage (allocs/op) |
185 |
16 |
-91.4% |
Full pipeline ProcessMessage (Avro) |
6.60 µs |
1.53 µs |
-76.8% |
Single-threaded full-pipeline throughput, no network I/O: 152k msg/s to ~654k msg/s (4.3x).
Fix #2: So was the Protobuf schema
Same shape of bug, different format. ParseProtobufSchema ran protocompile.Compiler.Compile(...), a full .proto compile, on every call.
The fix is the same LRU pattern, with the compile logic pulled into a small compileProtobufSchema helper so the cached and flag-disabled paths share one implementation:
var protobufSchemaCache = newLRUCache[*ParsedProtobufSchema](
cacheSizeFromEnv("KAFKA_DS_PERF_PROTOBUF_SCHEMA_CACHE_MAX_ENTRIES", 256),
)
func ParseProtobufSchema(schema string) (*ParsedProtobufSchema, error) {
if perfflags.ProtobufSchemaCache.Disabled() {
return compileProtobufSchema(schema)
}
if cached, ok := protobufSchemaCache.Get(schema); ok {
return cached, nil
}
parsed, err := compileProtobufSchema(schema)
if err != nil {
return nil, err
}
protobufSchemaCache.Add(schema, parsed)
return parsed, nil
}
Results:
| Benchmark |
Before |
After |
Change |
DecodeProtobufMessage_Plain (ns/op) |
15,653 ns |
574 ns |
-96.3% |
DecodeProtobufMessage_Plain (B/op) |
35,862 B |
997 B |
-97.2% |
DecodeProtobufMessage_Plain (allocs/op) |
250 |
14 |
-94.4% |
Full pipeline ProcessMessage (Protobuf) |
19.07 µs |
1.53 µs |
-92.0% |
Throughput: 52k msg/s to ~656k msg/s (12.5x). The biggest win in the whole investigation, and schema recompilation, not byte decoding, was almost the entire cost.
One detail worth highlighting because it's good practice: JSON and plaintext benchmarks share no code with the Avro/Protobuf decode path, and they were unaffected by both fixes. That's a control group. If they had moved, something about my measurement was wrong.
Fix #3: Re-sorting field names every message (a smaller, honest win)
The problem. For JSON, Avro, and Protobuf messages, ProcessMessage collects the flattened field keys into a slice and calls sort.Strings(keys) on every message, so Grafana's frame columns keep a stable order. If a topic's schema doesn't change between messages, which is the common case, that sort is redundant.
The fix. Cache the sorted order on the StreamManager and reuse it whenever the current message's key set matches the previous one:
func (sm *StreamManager) sortedFlatKeys(flat map[string]interface{}) []string {
if !perfflags.FieldOrderCache.Disabled() {
sm.mu.RLock()
hit := sm.flatKeySetMatchesLocked(flat)
cached := sm.flatKeyOrder
sm.mu.RUnlock()
if hit {
return cached
}
}
// cache miss (or flag disabled): collect + sort from scratch, then cache
...
}
Results, and here I'd rather be honest than sell a bigger win than exists:
| Field count |
Before |
After |
Change |
| 20 fields (typical topic) |
3.855 µs |
3.543 µs |
-8.1% (p=0.002) |
| 100 fields (wide topic) |
19.15 µs |
16.12 µs |
-15.8% (p=0.002) |
Eight percent is not a headline. But it's real, it's statistically significant, and it grows with schema width, since sorting cost rises faster than the cache-check overhead. Not every fix is a 12x. Most aren't.
Fix #4: One frame per message was the wrong unit
The three fixes above all say "stop doing redundant work." This one changes the unit of work instead, and it's the one I'd most want you to remember.
The plugin built one data frame per Kafka message and sent it immediately. Frames go out with data.IncludeAll, which means every send re-serializes the entire schema alongside a single row of data. At high message rates you're shipping the schema over and over, and for a wide topic the schema is far bigger than the row.
The fix is to accumulate compatible frames and send them as one multi-row frame:
What counts as "compatible"?
Two frames are safe to merge only if they'd produce identical columns. frameSchemaKey(frame) builds a string key that fingerprints exactly that, nothing more:
Frame name, RefID, and field count
A hash of frame.Meta, if the frame has any
Per field, in order: name:type, its labels, and a hash of FieldConfig
Frames whose keys match get merged into one (EmptyCopy() plus AppendRow(...) for each row). Frames whose keys differ stay in separate pending groups and flush separately, in the order they first appeared.
One thing this key deliberately leaves out: partition. Partition is a column value, not part of the schema, so two partitions of the same topic still batch together, which is what you want. The opposite case, a message whose field set changes mid-stream, simply opens a new group instead of merging into the wrong one.
What triggers a flush?
Whichever comes first:
Row cap. 32 rows, counted across all pending groups (defaultMicroBatchMaxRows). Hitting it flushes immediately.
Time window. 5 ms, on a ticker running alongside the message loop (defaultMicroBatchMaxLatency). Every tick flushes whatever's pending, so nothing waits longer than one tick.
Shutdown. ctx.Done() flushes before the stream closes, so no rows are lost on unsubscribe.
Two implementation details worth flagging if you're building something similar:
The merge step doesn't just trust the hash key. appendFrameRows re-checks field names, types, config, and labels directly before merging, and falls back to sending frames immediately on any mismatch. A bug in the key function can't silently corrupt a channel's schema.
Line Protocol is excluded from batching entirely, since one Kafka message there already expands into many rows. That exclusion turns out to be a free control group in the benchmark below.
Doesn't that add latency?
Yes, but it's capped at 5 ms, and shrinks as throughput rises. The two triggers cross over at 32 rows ÷ 5 ms = 6,400 msg/s. Below that rate, 32 rows take longer than 5 ms to arrive, so the ticker flushes first. Above it, 32 rows arrive in under 5 ms, so the row cap flushes first and the wait drops below 5 ms. Either way, that's an order of magnitude under Grafana Live's own panel repaint rate, so a live dashboard doesn't visibly burst.
Results (this fix in isolation)
These come from toggling only KAFKA_DS_PERF_DISABLE_STREAM_MICROBATCH, with the other three fixes left enabled, -count=6:
| Fixture |
ns/op before to after |
msg/s before to after |
Change |
plaintext |
1.443 µs to 1.031 µs |
693.3k to 970.2k |
-28.6% (p=0.002) |
avro |
2.670 µs to 2.076 µs |
374.7k to 481.8k |
-22.3% |
protobuf |
2.704 µs to 2.080 µs |
369.9k to 481.0k |
-23.1% |
json_20fields |
7.986 µs to 5.464 µs |
125.2k to 183.0k |
-31.6% |
json_100fields |
36.81 µs to 25.23 µs |
27.17k to 39.64k |
-31.5% |
lineprotocol (not batched) |
32.12 µs to 31.96 µs |
31.13k to 31.29k |
~ (p=0.093) |
| geomean |
6.793 µs to 5.192 µs |
147.2k to 192.6k |
-23.6% |
Line Protocol being flat is the sanity check. It's the one format that bypasses the batcher, and it's the one row that doesn't move.
But msg/s undersells what actually happened here. The real story is in packets/s:
| Fixture |
packets/s before to after |
Change |
plaintext |
693.3k to 30.3k |
-95.6% |
avro |
374.7k to 15.1k |
-96.0% |
protobuf |
369.9k to 15.0k |
-95.9% |
json_20fields |
125.2k to 5.72k |
-95.4% |
json_100fields |
27.17k to 1.24k |
-95.4% |
lineprotocol |
31.1k to 31.3k |
~ |
Divide msg/s by packets/s in any batched fixture and you get exactly 32.0, the row cap, which confirms the batcher is saturating rather than dribbling out half-full batches.
And the metric nobody asked for but that closes the argument: actual network throughput, bytes per second sent over the wire (not to be confused with the B/op allocation figures elsewhere in this post, that's memory, this is network). For json_100fields, throughput dropped from 262.75 MB/s to 36.50 MB/s (-86.1%), geomean -77.4% less data sent, for more messages delivered. That's schema amortization made visible: one schema shipped per 32 rows instead of 32 schemas for 32 rows.
The profile delta on json_100fields says the same thing from the CPU side:
CPU cumulative in data.FrameToJSON: 22.58% to 2.05%
alloc_objects cumulative in data.FrameToJSON: 14.79% to 6.44%
New costs that appear: AddFrames at 12.55%, appendFrameRows at 10.65%, Frame.RowCopy at 8.50% of alloc_objects (and only ~1.5%, ~0.5%, ~0.5% of CPU respectively)
Memory: 37.80 KiB to 34.95 KiB/op (-7.55%), but 833 to 869 allocs/op (+4.32%)
That last line is the trade-off stated plainly: fewer bytes, slightly more allocation calls, because copying rows into an aggregate frame isn't free. I'd rather show both columns than pick the flattering one.
Step 4: Prove it with benchstat
A single benchmark run tells you almost nothing. Go's benchmark output moves several percent between runs on the same binary. benchstat compares runs statistically and gives you a p-value:
go install golang.org/x/perf/cmd/benchstat@latest
# after (default, fixes on)
go test -bench='^BenchmarkDecodeAvroMessage$' -benchmem -run='^$' -count=6 \
./pkg/kafka_client/... > after.txt
# before (this one fix disabled)
KAFKA_DS_PERF_DISABLE_AVRO_CODEC_CACHE=true \
go test -bench='^BenchmarkDecodeAvroMessage$' -benchmem -run='^$' -count=6 \
./pkg/kafka_client/... > before.txt
benchstat before.txt after.txt
Two things matter here. -count=6 gives benchstat enough samples to say something. And use the identical -bench filter for both runs. If "before" includes benchmarks that didn't change, the aggregate comparison quietly becomes meaningless.
That's how I can claim the Fix #3 win at 8.1% with a straight face: p=0.002, not "it looked faster."
The trick that made all of this reproducible: feature flags
Here's the part I'd most encourage you to steal.
Every fix sits behind an environment-variable-backed flag in a pkg/perfflags package. By default the flags are unset, and you get the fast behavior. Set one, and the plugin reverts to exactly its pre-fix behavior. No git checkout, no second binary:
type Flag struct {
envVar string
value atomic.Bool
}
func (f *Flag) Disabled() bool { return f.value.Load() }
| Environment variable |
Reverts |
KAFKA_DS_PERF_DISABLE_AVRO_CODEC_CACHE |
Avro codec caching (Fix #1) |
KAFKA_DS_PERF_DISABLE_PROTOBUF_SCHEMA_CACHE |
Protobuf schema caching (Fix #2) |
KAFKA_DS_PERF_DISABLE_FIELD_ORDER_CACHE |
Field-order caching (Fix #3) |
KAFKA_DS_PERF_DISABLE_STREAM_MICROBATCH |
Micro-batching (Fix #4) |
This buys three things:
Verifiable claims. Every number in this post is something you can reproduce on your own machine in about a minute, by flipping one variable.
An operational escape hatch. If a cache misbehaves in someone's deployment, they set a variable and restart Grafana. They don't wait for me to cut a release.
Cheap benchmarks. No maintaining two branches just to compare against.
What I deliberately didn't fix
Profiling the JSON path showed FieldBuilder.AddValueToFrame as the largest remaining cost, around 40% of ProcessMessage's own time, bigger than flattening or sorting. Allocation profiling traced it to three allocations per field:
A make([]*float64, 1) pointer slice per field.
The Grafana SDK's internal nullable-vector wrapper, inside data.NewField.
Boxing the value into a pointer via SetConcrete, for nullable semantics.
I left it alone on purpose. The plugin always builds nullable field types, even for non-null values, so a field's Go type stays consistent across the live stream. Grafana Live requires stable schema across appended frames. Non-nullable slices would save an allocation per field and risk breaking that stability the first time a message needs to send a null.
That's a real risk I can't currently verify without an end-to-end Grafana Live test harness. A fix I can't prove safe is not a fix. It stays in the "next candidates" pile, along with a lock-free fast path for single-consumer streams, parallel decode workers with ordering guarantees, and singleflight de-duplication for cold schema-registry bursts.
Everything together
Combining all four fixes on the whole-workflow benchmark in sendframe_json mode (n=6):
| Case |
Before |
After |
Change |
plaintext |
662.1k |
886.0k |
+33.8% |
lineprotocol |
29.37k |
29.66k |
~0% |
avro |
122.4k |
457.0k |
+273.3% |
protobuf |
47.39k |
451.9k |
+853.5% |
json_20fields |
115.8k |
172.5k |
+49.0% |
json_100fields |
24.21k |
37.62k |
+55.4% |
| geomean |
82.55k |
181.0k |
+119.3% |
Geomean throughput a bit more than doubled. And lineprotocol didn't move at all, which is exactly right, since none of these fixes touch that path.
Conclusion
The headline number is 12x, but the headline number is the least interesting thing here. What I'd actually take away:
Benchmark = the number. Without it you have opinions.
Profile = the line. Function-level output is a hint, -list is the answer.
Fix = only what the data justifies. Two caches did almost all the work.
Prove = benchstat with -count=6. "Faster" means "faster beyond noise."
Feature-flag it = keep the old behavior one environment variable away, forever.
Your Turn
What's the most redundant work hiding on your hot path? Something recompiled, re-parsed, or re-sorted per request that could be computed once?
Is your unit of work the right size, or are you paying fixed overhead once per item when you could pay it once per batch?
When you last claimed something got faster, did you have a p-value, or one good run?
If you try the reproduction steps and get wildly different numbers on your hardware, I'd like to hear about it. The repo is here, and issues are open.
