Updating kfake fails TestManagerMetrics

[rubvs]

From main run

> go get -u github.com/twmb/franz-go/pkg/kfake
> go mod tidy
> go test -v -race -timeout=10s ./... -run TestManagerMetrics

=== RUN   TestManagerMetrics
    manager_test.go:431:
                Error Trace:    /Users/rubenvanstaden/issue/apm_queue/661/apm-queue/kafka/manager_test.go:431
                Error:          "[{messaging.kafka.connects.count Total number of connections opened, by broker 1 {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {outcome {4 0 success <nil>}}]}} 2025-05-28 19:58:21.380781 -0400 EDT m=+0.007719668 2025-05-28 19:58:21.386697 -0400 EDT m=+0.013636084 4 []}] CumulativeTemporality true}} {messaging.kafka.disconnects.count Total number of connections closed, by broker 1 {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}}]}} 2025-05-28 19:58:21.380814 -0400 EDT m=+0.007752459 2025-05-28 19:58:21.386709 -0400 EDT m=+0.013648043 2 []}] CumulativeTemporality true}} {messaging.kafka.write_bytes Total number of bytes written, by broker By {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 Metadata <nil>}}]}} 2025-05-28 19:58:21.38083 -0400 EDT m=+0.007768668 2025-05-28 19:58:21.386714 -0400 EDT m=+0.013652543 392 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 FindCoordinator <nil>}}]}} 2025-05-28 19:58:21.38083 -0400 EDT m=+0.007768668 2025-05-28 19:58:21.386714 -0400 EDT m=+0.013652543 90 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 DescribeGroups <nil>}}]}} 2025-05-28 19:58:21.38083 -0400 EDT m=+0.007768668 2025-05-28 19:58:21.386714 -0400 EDT m=+0.013652543 59 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 OffsetFetch <nil>}}]}} 2025-05-28 19:58:21.38083 -0400 EDT m=+0.007768668 2025-05-28 19:58:21.386714 -0400 EDT m=+0.013652543 104 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 ListOffsets <nil>}}]}} 2025-05-28 19:58:21.38083 -0400 EDT m=+0.007768668 2025-05-28 19:58:21.386714 -0400 EDT m=+0.013652543 390 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 ApiVersions <nil>}}]}} 2025-05-28 19:58:21.38083 -0400 EDT m=+0.007768668 2025-05-28 19:58:21.386714 -0400 EDT m=+0.013652543 124 []}] CumulativeTemporality true}} {messaging.kafka.read_bytes.count Total number of bytes read, by broker By {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}}]}} 2025-05-28 19:58:21.380845 -0400 EDT m=+0.007783501 2025-05-28 19:58:21.386733 -0400 EDT m=+0.013671876 3137 []}] CumulativeTemporality true}} {messaging.kafka.write.latency Time it took to write a batch including wait time before writing s {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 ApiVersions <nil>}} {outcome {4 0 success <nil>}}]}} 2025-05-28 19:58:21.380954 -0400 EDT m=+0.007892626 2025-05-28 19:58:21.38674 -0400 EDT m=+0.013679043 4 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 4 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {1.4707999999999999e-05 true} {9.2583e-05 true} 0.000154707 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 Metadata <nil>}} {outcome {4 0 success <nil>}}]}} 2025-05-28 19:58:21.380954 -0400 EDT m=+0.007892626 2025-05-28 19:58:21.38674 -0400 EDT m=+0.013679043 5 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 5 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {1.8125e-05 true} {0.000858875 true} 0.0010064170000000001 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 FindCoordinator <nil>}} {outcome {4 0 success <nil>}}]}} 2025-05-28 19:58:21.380954 -0400 EDT m=+0.007892626 2025-05-28 19:58:21.38674 -0400 EDT m=+0.013679043 2 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 2 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {6.9625e-05 true} {0.00064325 true} 0.0007128750000000001 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 DescribeGroups <nil>}} {outcome {4 0 success <nil>}}]}} 2025-05-28 19:58:21.380954 -0400 EDT m=+0.007892626 2025-05-28 19:58:21.38674 -0400 EDT m=+0.013679043 1 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {0.000430042 true} {0.000430042 true} 0.000430042 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 OffsetFetch <nil>}} {outcome {4 0 success <nil>}}]}} 2025-05-28 19:58:21.380954 -0400 EDT m=+0.007892626 2025-05-28 19:58:21.38674 -0400 EDT m=+0.013679043 1 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {4.675e-05 true} {4.675e-05 true} 4.675e-05 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 ListOffsets <nil>}} {outcome {4 0 success <nil>}}]}} 2025-05-28 19:58:21.380954 -0400 EDT m=+0.007892626 2025-05-28 19:58:21.38674 -0400 EDT m=+0.013679043 2 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 2 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {0.00036095800000000004 true} {0.000398209 true} 0.000759167 []}] CumulativeTemporality}} {messaging.kafka.read.latency Time it took to read a batch including wait time before reading s {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {outcome {4 0 success <nil>}}]}} 2025-05-28 19:58:21.380984 -0400 EDT m=+0.007923001 2025-05-28 19:58:21.386763 -0400 EDT m=+0.013701709 15 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 15 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {9.9541e-05 true} {0.00028233399999999997 true} 0.0022929179999999997 []}] CumulativeTemporality}}]" should have 8 item(s), but has 6
                Test:           TestManagerMetrics
--- FAIL: TestManagerMetrics (0.01s)

From should have 8 item(s), but has 6, the metrics count should be 8, but is 6, see Code.

[rubvs]

With upgrading franz-go and specifically kfake, it seems the cluster is unable to properly create the topics:

• Apply the following patch.

diff --git a/kafka/manager.go b/kafka/manager.go
index 98fb6f7..6f77411 100644
--- a/kafka/manager.go
+++ b/kafka/manager.go
@@ -275,6 +275,7 @@ func (m *Manager) MonitorConsumerLag(topicConsumers []apmqueue.TopicConsumer) (m
 				}
 				for partition, lag := range partitions {
 					if lag.Err != nil {
+						fmt.Println("lag.Err:", lag.Err, topic)
 						logger.Warn("error getting consumer group lag",
 							zap.String("group", l.Group),
 							zap.String("topic", topic),

• On the latest main in apm-queue with the applied patch.

> go test -v -failfast -race -timeout=10s ./... -run TestManagerMetrics

?       github.com/elastic/apm-queue/v2 [no test files]
?       github.com/elastic/apm-queue/v2/cmd/queuebench  [no test files]
=== RUN   TestManagerMetrics
lag.Err: missing from list offsets topic2
--- PASS: TestManagerMetrics (0.01s)
PASS
ok      github.com/elastic/apm-queue/v2/kafka   1.590s
?       github.com/elastic/apm-queue/v2/metrictest      [no test files]
testing: warning: no tests to run
PASS
ok      github.com/elastic/apm-queue/v2/queuecontext    (cached) [no tests to run]
2025-07-06T16:16:44.716-0400    INFO    systemtest/main_test.go:40      skipping system tests, pass -run-system-tests to enable
ok      github.com/elastic/apm-queue/v2/systemtest      1.312s

• Next, update franz-go and rerun the unit test

> go get -u github.com/twmb/franz-go/pkg/kfake
> go mod tidy
> go test -v -failfast -race -timeout=10s ./... -run TestManagerMetrics

?       github.com/elastic/apm-queue/v2 [no test files]
?       github.com/elastic/apm-queue/v2/cmd/queuebench  [no test files]
=== RUN   TestManagerMetrics
lag.Err: UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition. topic1
lag.Err: UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition. topic2
lag.Err: missing from list offsets topic2
lag.Err: UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition. topic3
lag.Err: UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition. mytopic
lag.Err: UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition. topic1
    manager_test.go:431:
                Error Trace:    /Users/rubenvanstaden/apm-queue/kafka/manager_test.go:431
                Error:          "[{messaging.kafka.connects.count Total number of connections opened, by broker 1 {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {outcome {4 0 success <nil>}}]}} 2025-07-06 16:18:33.125949 -0400 EDT m=+0.008062209 2025-07-06 16:18:33.132032 -0400 EDT m=+0.014145334 4 []}] CumulativeTemporality true}} {messaging.kafka.disconnects.count Total number of connections closed, by broker 1 {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}}]}} 2025-07-06 16:18:33.125984 -0400 EDT m=+0.008097001 2025-07-06 16:18:33.132054 -0400 EDT m=+0.014167459 2 []}] CumulativeTemporality true}} {messaging.kafka.write_bytes Total number of bytes written, by broker By {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 Metadata <nil>}}]}} 2025-07-06 16:18:33.125999 -0400 EDT m=+0.008112167 2025-07-06 16:18:33.132058 -0400 EDT m=+0.014171251 392 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 FindCoordinator <nil>}}]}} 2025-07-06 16:18:33.125999 -0400 EDT m=+0.008112167 2025-07-06 16:18:33.132058 -0400 EDT m=+0.014171251 90 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 DescribeGroups <nil>}}]}} 2025-07-06 16:18:33.125999 -0400 EDT m=+0.008112167 2025-07-06 16:18:33.132058 -0400 EDT m=+0.014171251 59 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 OffsetFetch <nil>}}]}} 2025-07-06 16:18:33.125999 -0400 EDT m=+0.008112167 2025-07-06 16:18:33.132058 -0400 EDT m=+0.014171251 104 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 ListOffsets <nil>}}]}} 2025-07-06 16:18:33.125999 -0400 EDT m=+0.008112167 2025-07-06 16:18:33.132058 -0400 EDT m=+0.014171251 390 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 ApiVersions <nil>}}]}} 2025-07-06 16:18:33.125999 -0400 EDT m=+0.008112167 2025-07-06 16:18:33.132058 -0400 EDT m=+0.014171251 124 []}] CumulativeTemporality true}} {messaging.kafka.read_bytes.count Total number of bytes read, by broker By {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}}]}} 2025-07-06 16:18:33.126012 -0400 EDT m=+0.008125834 2025-07-06 16:18:33.132073 -0400 EDT m=+0.014186042 3137 []}] CumulativeTemporality true}} {messaging.kafka.write.latency Time it took to write a batch including wait time before writing s {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 OffsetFetch <nil>}} {outcome {4 0 success <nil>}}]}} 2025-07-06 16:18:33.12611 -0400 EDT m=+0.008223126 2025-07-06 16:18:33.132082 -0400 EDT m=+0.014195167 1 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {3.3416e-05 true} {3.3416e-05 true} 3.3416e-05 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 ListOffsets <nil>}} {outcome {4 0 success <nil>}}]}} 2025-07-06 16:18:33.12611 -0400 EDT m=+0.008223126 2025-07-06 16:18:33.132082 -0400 EDT m=+0.014195167 2 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 2 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {0.0003575 true} {0.000367209 true} 0.000724709 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 ApiVersions <nil>}} {outcome {4 0 success <nil>}}]}} 2025-07-06 16:18:33.12611 -0400 EDT m=+0.008223126 2025-07-06 16:18:33.132082 -0400 EDT m=+0.014195167 4 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 4 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {1.5166999999999999e-05 true} {0.000134792 true} 0.00018554200000000002 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 Metadata <nil>}} {outcome {4 0 success <nil>}}]}} 2025-07-06 16:18:33.12611 -0400 EDT m=+0.008223126 2025-07-06 16:18:33.132082 -0400 EDT m=+0.014195167 5 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 5 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {2.0249999999999998e-05 true} {0.0010135839999999999 true} 0.00115025 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 FindCoordinator <nil>}} {outcome {4 0 success <nil>}}]}} 2025-07-06 16:18:33.12611 -0400 EDT m=+0.008223126 2025-07-06 16:18:33.132082 -0400 EDT m=+0.014195167 2 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 2 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {5.5708e-05 true} {0.000809708 true} 0.000865416 []} {{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {operation {4 0 DescribeGroups <nil>}} {outcome {4 0 success <nil>}}]}} 2025-07-06 16:18:33.12611 -0400 EDT m=+0.008223126 2025-07-06 16:18:33.132082 -0400 EDT m=+0.014195167 1 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {0.0004325 true} {0.0004325 true} 0.0004325 []}] CumulativeTemporality}} {messaging.kafka.read.latency Time it took to read a batch including wait time before reading s {[{{{[{messaging.system {4 0 kafka <nil>}} {namespace {4 0 name_space <nil>}} {outcome {4 0 success <nil>}}]}} 2025-07-06 16:18:33.126149 -0400 EDT m=+0.008262042 2025-07-06 16:18:33.132104 -0400 EDT m=+0.014217876 15 [0 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500 10000] [0 15 0 0 0 0 0 0 0 0 0 0 0 0 0 0] {0.000107542 true} {0.000326583 true} 0.0022866270000000003 []}] CumulativeTemporality}}]" should have 8 item(s), but has 6
                Test:           TestManagerMetrics
--- FAIL: TestManagerMetrics (0.01s)
FAIL
FAIL    github.com/elastic/apm-queue/v2/kafka   0.360s
?       github.com/elastic/apm-queue/v2/metrictest      [no test files]
testing: warning: no tests to run
PASS
ok      github.com/elastic/apm-queue/v2/queuecontext    (cached) [no tests to run]
2025-07-06T16:18:33.272-0400    INFO    systemtest/main_test.go:40      skipping system tests, pass -run-system-tests to enable
ok      github.com/elastic/apm-queue/v2/systemtest      1.507s
FAIL

[rubvs]

Created an issue in franz-go with a a minimal reproduce: twmb/franz-go#1056

[simitt]

@rubvs in the issue you created you showed that there was also an error before the upgrade; a slightly different one but also indicating topic creation issue/topic missing. Can you add to why this wasn't a problem before but is now?

ERR: missing from list offsets topic: topic1, partition: 1, lag: -1

vs now

ERR: UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition. topic: topic1, partition: 1, lag: -1

[rubvs]

@simitt

• The changes relate to this PR in franz-go: kadm: improve Lag calculation to account for log-start-offset being non-zero twmb/franz-go#744.
• The errListMissing error message also show up in our current version of apm-queue, but we tend to ignore it.
• However, with the PR in franz-go listed above, the UnknownTopicOrPartition message is propagated instead of errListMissing which is handled differently within the codebase, leading to the unit tests failing eventually.

All fixes has been aggregated in my latest PR: #675