kafka 2.2 low_water_mark is the same as high

[hdhoang]

Hello team, thanks for the excellent exporter

We have some clusters with confluent kafka 2.2 (platform 5.2). After changing to kminion 2.2.11, the topic/partition low_water_mark values are mostly identical to the high values, thus message counting are wrong.

Reproduction

Based on docker-compose:

1. change to cp-kafka 5.2.7

kafka:
  image: confluentinc/cp-kafka:5.2.7

2. podman-compose up (sorry, I'm on fedora)
3. start v2.2.10, v2.2.11 in the same network

podman run --network kminion_default -p 8210:8080 -e KAFKA_BROKERS=kafka:29092  redpandadata/kminion:v2.2.10
podman run --network kminion_default -p 8211:8080 -e KAFKA_BROKERS=kafka:29092  redpandadata/kminion:v2.2.11

4. observe low_water_mark difference for the automatically created topic. port 8080 with latest commit is like v2.2.11

$ diff -u2 <(curl -s 127.1:8211/metrics) <(curl -s 127.1:8210/metrics) | grep -E 'info|version|water'
 # HELP go_info Information about the Go environment.
 # TYPE go_info gauge
-go_info{version="go1.24.1"} 1
+go_info{version="go1.22.6"} 1
-# HELP go_memstats_mallocs_total Total number of heap objects allocated, both live and gc-ed. Semantically a counter version for go_memstats_heap_objects gauge. Equals to /gc/heap/allocs:objects + /gc/heap/tiny/allocs:objects.
 # HELP kminion_kafka_topic_high_water_mark_sum Sum of all the topic's partition high water marks
 # TYPE kminion_kafka_topic_high_water_mark_sum gauge
 # HELP kminion_kafka_topic_low_water_mark_sum Sum of all the topic's partition low water marks
 # TYPE kminion_kafka_topic_low_water_mark_sum gauge
-kminion_kafka_topic_low_water_mark_sum{topic_name="__confluent.support.metrics"} 1
+kminion_kafka_topic_low_water_mark_sum{topic_name="__confluent.support.metrics"} 0
 # HELP kminion_kafka_topic_partition_high_water_mark Partition High Water Mark
 # TYPE kminion_kafka_topic_partition_high_water_mark gauge
 # HELP kminion_kafka_topic_partition_low_water_mark Partition Low Water Mark
 # TYPE kminion_kafka_topic_partition_low_water_mark gauge
-kminion_kafka_topic_partition_low_water_mark{partition_id="0",topic_name="__confluent.support.metrics"} 1
+kminion_kafka_topic_partition_low_water_mark{partition_id="0",topic_name="__confluent.support.metrics"} 0

v2.2.10 debug without ts

{"level":"debug","broker":"1","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kafka:29092"}
{"level":"debug","dial_duration":0.001736102,"logger":"main.kafka_hooks","msg":"kafka connection succeeded","host":"kafka"}
{"level":"debug","broker":"1","logger":"main.kgo_client","msg":"connection opened to broker","addr":"kafka:29092"}
{"level":"debug","broker":"1","logger":"main.kgo_client","msg":"connection initialized successfully","addr":"kafka:29092"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote ApiVersions v2","broker":"1","bytes_written":21,"write_wait":0.001974268,"time_to_write":0.000041603}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read ApiVersions v2","broker":"1","bytes_read":282,"read_wait":0.000096094,"time_to_read":0.005220454}
{"level":"debug","broker":"seed_0","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kafka:29092"}
{"level":"debug","dial_duration":0.001710923,"logger":"main.kafka_hooks","msg":"kafka connection succeeded","host":"kafka"}
{"level":"debug","broker":"seed_0","logger":"main.kgo_client","msg":"connection opened to broker","addr":"kafka:29092"}
{"level":"debug","broker":"seed_0","logger":"main.kgo_client","msg":"connection initialized successfully","addr":"kafka:29092"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"seed_0","bytes_written":26,"write_wait":0.002236381,"time_to_write":0.000067513}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"seed_0","bytes_read":135,"read_wait":0.000088153,"time_to_read":0.005245559}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"1","bytes_written":26,"write_wait":0.000226303,"time_to_write":0.00014392}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"1","bytes_read":135,"read_wait":0.000283829,"time_to_read":0.002441513}
{"level":"debug","destinations":["1"],"logger":"main.kgo_client","msg":"sharded request","req":"DescribeLogDirs"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote DescribeLogDirs v1","broker":"1","bytes_written":25,"write_wait":0.00005342,"time_to_write":0.000069947}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read DescribeLogDirs v1","broker":"1","bytes_read":97,"read_wait":0.00009336,"time_to_read":0.005681271}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"seed_0","bytes_written":26,"write_wait":0.000024999,"time_to_write":0.000060346}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"seed_0","bytes_read":135,"read_wait":0.000064304,"time_to_read":0.002519752}
{"level":"debug","destinations":["1"],"logger":"main.kgo_client","msg":"sharded request","req":"ListGroups"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote ListGroups v2","broker":"1","bytes_written":21,"write_wait":0.000074024,"time_to_write":0.000072735}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read ListGroups v2","broker":"1","bytes_read":18,"read_wait":0.000136295,"time_to_read":0.001150263}
{"level":"debug","destinations":["any"],"logger":"main.kgo_client","msg":"sharded request","req":"DescribeGroups"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote DescribeGroups v2","broker":"seed_0","bytes_written":25,"write_wait":0.000175722,"time_to_write":0.0001768}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read DescribeGroups v2","broker":"seed_0","bytes_read":16,"read_wait":0.0001665,"time_to_read":0.000748145}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"1","bytes_written":55,"write_wait":0.000052055,"time_to_write":0.000818636}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"1","bytes_read":135,"read_wait":0.000245011,"time_to_read":0.002224917}
{"level":"debug","destinations":["1"],"logger":"main.kgo_client","msg":"sharded request","req":"ListOffsets"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote ListOffsets v5","broker":"1","bytes_written":79,"write_wait":0.000035976,"time_to_write":0.000050171}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read ListOffsets v5","broker":"1","bytes_read":75,"read_wait":0.000083177,"time_to_read":0.00266406}
{"level":"debug","destinations":["1"],"logger":"main.kgo_client","msg":"sharded request","req":"ListOffsets"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote ListOffsets v5","broker":"1","bytes_written":79,"write_wait":0.000064705,"time_to_write":0.000070259}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read ListOffsets v5","broker":"1","bytes_read":75,"read_wait":0.000561318,"time_to_read":0.000712053}
{"level":"debug","destinations":["any"],"logger":"main.kgo_client","msg":"sharded request","req":"DescribeConfigs"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote DescribeConfigs v2","broker":"1","bytes_written":60,"write_wait":0.000035928,"time_to_write":0.000060256}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read DescribeConfigs v2","broker":"1","bytes_read":1005,"read_wait":0.000244164,"time_to_read":0.005174959}

v2.2.11 debug without ts

{"level":"debug","broker":"1","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kafka:29092"}
{"level":"debug","dial_duration":0.003488341,"logger":"main.kafka_hooks","msg":"kafka connection succeeded","host":"kafka"}
{"level":"debug","broker":"1","logger":"main.kgo_client","msg":"connection opened to broker","addr":"kafka:29092"}
{"level":"debug","broker":"1","logger":"main.kgo_client","msg":"connection initialized successfully","addr":"kafka:29092"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote ApiVersions v2","broker":"1","bytes_written":21,"write_wait":0.004078185,"time_to_write":0.00008222}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read ApiVersions v2","broker":"1","bytes_read":282,"read_wait":0.000084797,"time_to_read":0.003153021}
{"level":"debug","broker":"seed_0","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kafka:29092"}
{"level":"debug","dial_duration":0.001358802,"logger":"main.kafka_hooks","msg":"kafka connection succeeded","host":"kafka"}
{"level":"debug","broker":"seed_0","logger":"main.kgo_client","msg":"connection opened to broker","addr":"kafka:29092"}
{"level":"debug","broker":"seed_0","logger":"main.kgo_client","msg":"connection initialized successfully","addr":"kafka:29092"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"seed_0","bytes_written":26,"write_wait":0.002190247,"time_to_write":0.000076989}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"seed_0","bytes_read":135,"read_wait":0.000122212,"time_to_read":0.00606761}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"1","bytes_written":26,"write_wait":0.000039645,"time_to_write":0.00010773}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"1","bytes_read":135,"read_wait":0.000110816,"time_to_read":0.003744385}
{"level":"debug","destinations":["1"],"logger":"main.kgo_client","msg":"sharded request","req":"DescribeLogDirs"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote DescribeLogDirs v1","broker":"1","bytes_written":25,"write_wait":0.000067479,"time_to_write":0.000084572}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read DescribeLogDirs v1","broker":"1","bytes_read":97,"read_wait":0.000112228,"time_to_read":0.003512425}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"seed_0","bytes_written":26,"write_wait":0.000033229,"time_to_write":0.000728576}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"seed_0","bytes_read":135,"read_wait":0.000173902,"time_to_read":0.001155734}
{"level":"debug","destinations":["1"],"logger":"main.kgo_client","msg":"sharded request","req":"ListGroups"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote ListGroups v2","broker":"1","bytes_written":21,"write_wait":0.00002631,"time_to_write":0.000041269}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read ListGroups v2","broker":"1","bytes_read":18,"read_wait":0.000059436,"time_to_read":0.001092918}
{"level":"debug","destinations":["any"],"logger":"main.kgo_client","msg":"sharded request","req":"DescribeGroups"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote DescribeGroups v2","broker":"seed_0","bytes_written":25,"write_wait":0.000037455,"time_to_write":0.000047764}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read DescribeGroups v2","broker":"seed_0","bytes_read":16,"read_wait":0.000076783,"time_to_read":0.000630923}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"1","bytes_written":26,"write_wait":0.000027897,"time_to_write":0.000035072}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"1","bytes_read":135,"read_wait":0.000081841,"time_to_read":0.001729692}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"1","bytes_written":55,"write_wait":0.000026574,"time_to_write":0.000049128}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"1","bytes_read":135,"read_wait":0.000121947,"time_to_read":0.001061557}
{"level":"debug","destinations":["1"],"logger":"main.kgo_client","msg":"sharded request","req":"ListOffsets"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote ListOffsets v5","broker":"1","bytes_written":79,"write_wait":0.000027254,"time_to_write":0.000066685}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read ListOffsets v5","broker":"1","bytes_read":75,"read_wait":0.000069554,"time_to_read":0.001223573}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote Metadata v7","broker":"seed_0","bytes_written":26,"write_wait":0.00005299,"time_to_write":0.000058656}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read Metadata v7","broker":"seed_0","bytes_read":135,"read_wait":0.000259466,"time_to_read":0.002189329}
{"level":"debug","destinations":["1"],"logger":"main.kgo_client","msg":"sharded request","req":"ListOffsets"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote ListOffsets v5","broker":"1","bytes_written":79,"write_wait":0.000074443,"time_to_write":0.000129866}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read ListOffsets v5","broker":"1","bytes_read":75,"read_wait":0.000301541,"time_to_read":0.00175023}
{"level":"debug","destinations":["any"],"logger":"main.kgo_client","msg":"sharded request","req":"DescribeConfigs"}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"wrote DescribeConfigs v2","broker":"seed_0","bytes_written":60,"write_wait":0.000037525,"time_to_write":0.000060102}
{"level":"debug","err":null,"logger":"main.kgo_client","msg":"read DescribeConfigs v2","broker":"seed_0","bytes_read":1005,"read_wait":0.000102147,"time_to_read":0.00584207}

Suggestion

I suppose something in #277 changed return values?

[someonestolemysweetroll]

Hi everyone! any idea when the fix will be released? we're seeing the same issue on 2.2.13

[korcadr]

same as #302