mirror of
https://github.com/hanzoai/kafka.git
synced 2026-10-30 14:48:25 +00:00
zeekayandHanzo Dev
0c384b11ba
A fetch with nothing to read read both partition edges and then slept
a fixed 300 ms whatever the client asked for. Sixty caught-up
consumers kept the broker and the store busy with no record moving.
Bounds are now kept per partition stream with the state they were read
from (stream created time, first and last sequence, message count). A
read whose StreamInfo reports the same state answers from them, so an
unchanged partition costs one StreamInfo and reads no message. Bounds
read through a failed or overtaken read serve that call and are not
kept.
Fetch answers the way Kafka's delayed fetch does: at once on MinBytes
or any partition error, otherwise when a requested partition gets a
new tail or MaxWaitMs runs out, capped at 30 s. A fetch serves one
record set per partition, so one that leaves records behind answers at
once too: waiting could not bring it closer to MinBytes. Each produce announces
its partition's new tail on a core subject after the store acks it, so
a fetch parked on one broker wakes for a produce served by any broker
on the bus. Shutdown answers parked fetches.
BenchmarkIdleFetch on main (9a065af) and on this commit: 60 caught-up
consumers over 20 topics, Fetch v12, MaxWaitMs 500, three runs of 1200
fetches each:
cores JetStream API/s consumers/s fetch/s
before 0.48-0.53 887-916 177-183 177-183
after 0.056-0.067 119 0 119
Before, each fetch made five API requests (stream info, stream names,
consumer create, consumer delete, message get); after, one stream info.
Co-authored-by: Hanzo Dev <[email protected]>
95 lines
2.6 KiB
Go
95 lines
2.6 KiB
Go
package protocol
|
|
|
|
import (
|
|
"fmt"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/twmb/franz-go/pkg/kmsg"
|
|
)
|
|
|
|
// BenchmarkIdleFetch measures what caught-up consumers cost when there is
|
|
// nothing to read: 60 connections over 20 topics, each looping the Fetch a
|
|
// librdkafka consumer sends when caught up (v12, the highest this broker
|
|
// serves, with MaxWaitMs 500 and MinBytes 1) against an embedded pubsub. One
|
|
// op is one fetch round trip. It reports process CPU per wall second in cores
|
|
// (broker, pubsub and clients together, as they share the process),
|
|
// JetStream API requests per second, and JetStream consumers created per
|
|
// second.
|
|
func BenchmarkIdleFetch(b *testing.B) {
|
|
const consumers, topics, records = 60, 20, 3
|
|
s := newStack(b)
|
|
for i := range topics {
|
|
name := fmt.Sprintf("idle-%02d", i)
|
|
s.topic(b, name, 1)
|
|
produce(b, s.addr, name, 0, records)
|
|
}
|
|
wires := make([]*wire, consumers)
|
|
reqs := make([]*kmsg.FetchRequest, consumers)
|
|
for i := range consumers {
|
|
wires[i] = dial(b, s.addr)
|
|
reqs[i] = fetchRequest(500, 1, at{fmt.Sprintf("idle-%02d", i%topics), 0, records})
|
|
}
|
|
failed := make(chan error, consumers)
|
|
// One fetch per connection first, so the measurement starts from open
|
|
// connections and partitions already read once.
|
|
var warm sync.WaitGroup
|
|
for i := range consumers {
|
|
warm.Go(func() {
|
|
if _, err := wires[i].fetch(12, fetchRequest(0, 1, at{fmt.Sprintf("idle-%02d", i%topics), 0, records})); err != nil {
|
|
failed <- err
|
|
}
|
|
})
|
|
}
|
|
warm.Wait()
|
|
select {
|
|
case err := <-failed:
|
|
b.Fatal(err)
|
|
default:
|
|
}
|
|
api := s.watchAPI(b)
|
|
|
|
var done atomic.Int64
|
|
var wg sync.WaitGroup
|
|
cpu0, t0 := cpuTime(b), time.Now()
|
|
b.ResetTimer()
|
|
for i := range consumers {
|
|
wg.Go(func() {
|
|
for done.Add(1) <= int64(b.N) {
|
|
resp, err := wires[i].fetch(12, reqs[i])
|
|
if err != nil {
|
|
failed <- err
|
|
return
|
|
}
|
|
p := resp.Topics[0].Partitions[0]
|
|
if p.ErrorCode != 0 || p.HighWatermark != records || len(p.RecordBatches) != 0 {
|
|
failed <- fmt.Errorf("caught-up fetch: error %d, high watermark %d, %d record bytes",
|
|
p.ErrorCode, p.HighWatermark, len(p.RecordBatches))
|
|
return
|
|
}
|
|
}
|
|
})
|
|
}
|
|
wg.Wait()
|
|
b.StopTimer()
|
|
wall, cpu := time.Since(t0), cpuTime(b)-cpu0
|
|
select {
|
|
case err := <-failed:
|
|
b.Fatal(err)
|
|
default:
|
|
}
|
|
ops, calls := api.take(b), 0
|
|
for _, n := range ops {
|
|
calls += n
|
|
}
|
|
|
|
sec := wall.Seconds()
|
|
b.ReportMetric(float64(b.N)/sec, "fetch/s")
|
|
b.ReportMetric(cpu.Seconds()/sec, "cores")
|
|
b.ReportMetric(float64(calls)/sec, "jsapi/s")
|
|
b.ReportMetric(float64(ops["CONSUMER.CREATE"])/sec, "consumers/s")
|
|
b.Logf("JetStream API requests over %v: %v", wall.Round(time.Millisecond), ops)
|
|
}
|