Repository: vectordotdev/vector (language: rust)

Issue:
Kafka lag metrics gives incorrect value
### A note for the community

<!-- Please keep this note for the community -->
* Please vote on this issue by adding a 👍 [reaction](https://blog.github.com/2016-03-10-add-reactions-to-pull-requests-issues-and-comments/) to the original issue to help the community and maintainers prioritize this request
* If you are interested in working on this issue or have submitted a pull request, please leave a comment
<!-- Thank you for keeping this note for the community -->


### Problem

I am working on vector dashboards and I noticed that kafka lag at some point spiked up probably due to some kafka or Elasticsearch glitch. But everything is processing just fine, it is also not growing, just settled on different level. And I don't see any delay in logs delivery either. Also digged deeper into per-partition metrics and it is similar for all (I wanted to make sure it is not just some partitions being stucked).

This chart shows amount of time (given current processing rate) needed to process all unconsumed messages. It would mean we have 1 hour delay which is not true, we have less than 5 minutes.
When I restarted one of vectors, it went back down to 5 minutes level.

<img width="1475" alt="image" src="https://github.com/user-attachments/assets/a4b81943-0092-428b-8382-76708ff29a80">

I tried to confirm it is vector/rdkafka issue by matching with kafka-lag-exporter metrics and it shows correct value.


Another interesting thing I found is metrics for partition_id `-1` being a negative value 😯
<img width="1455" alt="image" src="https://github.com/user-attachments/assets/0e383d22-0783-426b-833b-2dfebfcad713">


### Configuration

_No response_

### Version

0.39.0

### Debug Output

_No response_

### Example Data

_No response_

### Additional Context

_No response_

### References

_No response_


Discussion / hints:
Interesting, thanks for this report @fpytloun. Given the values don't match up with kafka-lag-exporter it does seem like a potential bug in Vector or rust-rdkafka or librdkafka.

Vector just publishes the metrics returned by rust-rdkafka: https://github.com/vectordotdev/vector/blob/fe2cc26a217364d5dd3f8c00289ce45af2446f24/src/internal_events/kafka.rs#L119-L152

I wonder if the measurement librdkafka is using is different than what kafka-lag-exporter measures? librdkafka documents it as:

> Difference between (hi_offset or ls_offset) and committed_offset). hi_offset is used when isolation.level=read_uncommitted, otherwise ls_offset.

https://github.com/confluentinc/librdkafka/blob/master/STATISTICS.md

If would be helpful, I could add a log message around here to log the data coming from librdkafka: https://github.com/vectordotdev/vector/blob/fe2cc26a217364d5dd3f8c00289ce45af2446f24/src/kafka.rs#L175-L185

Vector just ends up using the return values to emit metrics:

https://github.com/vectordotdev/vector/blob/master/src/internal_events/kafka.rs#L119-L152

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index f2a2e534c4774..55f2fd3a6f92f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6321,15 +6321,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "num_enum"
-version = "0.5.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1f646caf906c20226733ed5b1374287eb97e3c2a5c227ce668c1f2ce20ae57c9"
-dependencies = [
- "num_enum_derive 0.5.11",
-]
-
 [[package]]
 name = "num_enum"
 version = "0.6.1"
@@ -6348,18 +6339,6 @@ dependencies = [
  "num_enum_derive 0.7.3",
 ]
 
-[[package]]
-name = "num_enum_derive"
-version = "0.5.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dcbff9bc912032c62bf65ef1d5aea88983b420f4f839db1e9b0c281a25c9c799"
-dependencies = [
- "proc-macro-crate 1.3.1",
- "proc-macro2 1.0.92",
- "quote 1.0.37",
- "syn 1.0.109",
-]
-
 [[package]]
 name = "num_enum_derive"
 version = "0.6.1"
@@ -7904,9 +7883,9 @@ dependencies = [
 
 [[package]]
 name = "rdkafka"
-version = "0.35.0"
+version = "0.37.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f16c17f411935214a5870e40aff9291f8b40a73e97bf8de29e5959c473d5ef33"
+checksum = "14b52c81ac3cac39c9639b95c20452076e74b8d9a71bc6fc4d83407af2ea6fff"
 dependencies = [
  "futures-channel",
  "futures-util",
@@ -7922,15 +7901,15 @@ dependencies = [
 
 [[package]]
 name = "rdkafka-sys"
-version = "4.7.0+2.3.0"
+version = "4.8.0+2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "55e0d2f9ba6253f6ec72385e453294f8618e9e15c2c6aba2a5c01ccf9622d615"
+checksum = "ced38182dc436b3d9df0c77976f37a67134df26b050df1f0006688e46fc4c8be"
 dependencies = [
  "cmake",
  "curl-sys",
  "libc",
  "libz-sys",
- "num_enum 0.5.11",
+ "num_enum 0.7.3",
  "openssl-sys",
  "pkg-config",
  "sasl2-sys",
diff --git a/Cargo.toml b/Cargo.toml
index 12bb8fe919ca6..5df9dce099a04 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -350,7 +350,7 @@ postgres-openssl = { version = "0.5.0", default-features = false, features = ["r
 pulsar = { version = "6.3.0", default-features = false, features = ["tokio-runtime", "auth-oauth2", "flate2", "lz4", "snap", "zstd"], optional = true }
 rand.workspace = true
 rand_distr = { version = "0.4.3", default-features = false }
-rdkafka = { version = "0.35.0", default-features = false, features = ["curl-static", "tokio", "libz", "ssl", "zstd"], optional = true }
+rdkafka = { version = "0.37.0", default-features = false, features = ["curl-static", "tokio", "libz", "ssl", "zstd"], optional = true }
 redis = { version = "0.24.0", default-features = false, features = ["connection-manager", "tokio-comp", "tokio-native-tls-comp"], optional = true }
 regex = { version = "1.11.1", default-features = false, features = ["std", "perf"] }
 roaring = { version = "0.10.7", default-features = false, features = ["std"], optional = true }
diff --git a/src/sources/kafka.rs b/src/sources/kafka.rs
index bfa35e11924e1..7d0c1cdf26904 100644
--- a/src/sources/kafka.rs
+++ b/src/sources/kafka.rs
@@ -16,8 +16,8 @@ use futures::{Stream, StreamExt};
 use futures_util::future::OptionFuture;
 use rdkafka::{
     consumer::{
-        stream_consumer::StreamPartitionQueue, CommitMode, Consumer, ConsumerContext, Rebalance,
-        StreamConsumer,
+        stream_consumer::StreamPartitionQueue, BaseConsumer, CommitMode, Consumer, ConsumerContext,
+        Rebalance, StreamConsumer,
     },
     error::KafkaError,
     message::{BorrowedMessage, Headers as _, Message},
@@ -1369,15 +1369,11 @@ impl ClientContext for KafkaSourceContext {
 }
 
 impl ConsumerContext for KafkaSourceContext {
-    fn pre_rebalance(&self, rebalance: &Rebalance) {
+    fn pre_rebalance(&self, _base_consumer: &BaseConsumer<Self>, rebalance: &Rebalance) {
         match rebalance {
             Rebalance::Assign(tpl) => self.consume_partitions(tpl),
 
             Rebalance::Revoke(tpl) => {
-                // TODO  workaround for https://github.com/fede1024/rust-rdkafka/issues/681
-                if tpl.capacity() == 0 {
-                    return;
-                }
                 self.revoke_partitions(tpl);
                 self.commit_consumer_state();
             }


Test patch: