You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sahildevgon (via GitHub)" <gi...@apache.org> on 2026/06/02 14:35:16 UTC
[PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
sahildevgon opened a new pull request, #22454:
URL: https://github.com/apache/kafka/pull/22454
Extend the existing I/O metrics collection with the 5 missing metrics (rchar, wchar, syscr, syscw, and cancelled_write_bytes) that are essential for diagnosing performance issues and optimising Kafka deployments.
Details in the KIP below
https://cwiki.apache.org/confluence/x/co48G
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on PR #22454:
URL: https://github.com/apache/kafka/pull/22454#issuecomment-4668745109
Hi @m1a2st , I am unable to find that section, if you can share me the link, ill update the same there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3459952737
##########
server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java:
##########
@@ -79,32 +141,61 @@ public long writeBytes() {
* read_bytes: 0
* write_bytes: 0
* cancelled_write_bytes: 0
+ *
+ * <p>Parses into local variables first, then publishes to the volatile cached fields only
+ * after parsing succeeds. This prevents lock-free readers from observing the transient
+ * -1L state during an update. {@code lastUpdateMs} is written last so any thread observing
+ * the new timestamp via a volatile read also observes all updated cached fields.
+ *
+ * <p>Caller must hold the monitor of this instance.
*/
private boolean updateValues(long now) {
- synchronized (this) {
- try {
- cachedReadBytes = -1L;
- cachedWriteBytes = -1L;
- List<String> lines = Files.readAllLines(path, StandardCharsets.UTF_8);
- for (String line : lines) {
- if (line.startsWith(READ_BYTES_PREFIX)) {
- cachedReadBytes = Long.parseLong(line.substring(READ_BYTES_PREFIX.length()));
- } else if (line.startsWith(WRITE_BYTES_PREFIX)) {
- cachedWriteBytes = Long.parseLong(line.substring(WRITE_BYTES_PREFIX.length()));
- }
+ long readBytes = -1L;
+ long writeBytes = -1L;
+ long rchar = -1L;
+ long wchar = -1L;
+ long syscr = -1L;
+ long syscw = -1L;
+ long cancelledWriteBytes = -1L;
+ try {
+ List<String> lines = Files.readAllLines(path, StandardCharsets.UTF_8);
+ for (String line : lines) {
+ if (line.startsWith(READ_BYTES_PREFIX)) {
+ readBytes = Long.parseLong(line.substring(READ_BYTES_PREFIX.length()));
+ } else if (line.startsWith(WRITE_BYTES_PREFIX)) {
+ writeBytes = Long.parseLong(line.substring(WRITE_BYTES_PREFIX.length()));
+ } else if (line.startsWith(RCHAR_PREFIX)) {
+ rchar = Long.parseLong(line.substring(RCHAR_PREFIX.length()));
+ } else if (line.startsWith(WCHAR_PREFIX)) {
+ wchar = Long.parseLong(line.substring(WCHAR_PREFIX.length()));
+ } else if (line.startsWith(SYSCR_PREFIX)) {
+ syscr = Long.parseLong(line.substring(SYSCR_PREFIX.length()));
+ } else if (line.startsWith(SYSCW_PREFIX)) {
+ syscw = Long.parseLong(line.substring(SYSCW_PREFIX.length()));
+ } else if (line.startsWith(CANCELLED_WRITE_BYTES_PREFIX)) {
+ cancelledWriteBytes = Long.parseLong(line.substring(CANCELLED_WRITE_BYTES_PREFIX.length()));
}
- lastUpdateMs = now;
- return true;
- } catch (Throwable t) {
- LOG.warn("Unable to update IO metrics", t);
- return false;
}
+ } catch (Throwable t) {
+ LOG.warn("Unable to update IO metrics", t);
Review Comment:
Should we update `lastUpdateMs` here to avoid flooding the log?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3403087006
##########
core/src/main/scala/kafka/server/KafkaBroker.scala:
##########
@@ -120,5 +120,10 @@ trait KafkaBroker extends Logging {
if (linuxIoMetricsCollector.usable()) {
metricsGroup.newGauge("linux-disk-read-bytes", () => linuxIoMetricsCollector.readBytes())
Review Comment:
Could `linuxIoMetricsCollector` have a method to update them all at once instead of one by one? Otherwise, it will take the lock and access the underlying files frequently
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3485574442
##########
core/src/main/scala/kafka/server/ControllerServer.scala:
##########
@@ -145,6 +145,11 @@ class ControllerServer(
if (linuxIoMetricsCollector.usable()) {
Review Comment:
Added LinuxIoMetricsCollector.registerMetrics(KafkaMetricsGroup) that registers all 7 gauges via method references. Both KafkaBroker and ControllerServer now call it as a single line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3411479906
##########
server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java:
##########
@@ -79,32 +141,51 @@ public long writeBytes() {
* read_bytes: 0
* write_bytes: 0
* cancelled_write_bytes: 0
+ *
+ * <p>Caller must hold the monitor of this instance. {@code lastUpdateMs} is written last so
+ * that any thread observing the new timestamp via a volatile read also observes all updated
+ * cached fields.
*/
private boolean updateValues(long now) {
- synchronized (this) {
- try {
- cachedReadBytes = -1L;
- cachedWriteBytes = -1L;
- List<String> lines = Files.readAllLines(path, StandardCharsets.UTF_8);
- for (String line : lines) {
- if (line.startsWith(READ_BYTES_PREFIX)) {
- cachedReadBytes = Long.parseLong(line.substring(READ_BYTES_PREFIX.length()));
- } else if (line.startsWith(WRITE_BYTES_PREFIX)) {
- cachedWriteBytes = Long.parseLong(line.substring(WRITE_BYTES_PREFIX.length()));
- }
+ try {
+ cachedReadBytes = -1L;
Review Comment:
Hi @chia7712 , makes sense, pushed a patch where updateValues now parses into local variables and only publishes to the volatile cached fields once parsing succeeds, with lastUpdateMs written last so the
visibility ordering across all 7 fields plus the timestamp is consistent. Lock-free readers no longer see the transient -1L state.
One small behavior change , on parse failure (catch block), I now leave the cached fields untouched and return false, instead of setting them all to -1L. This way a transient /proc/self/io read failure doesn't make every JMX dashboard suddenly read -1L - readers retain last-known-good values. Happy to revert to the old "publish -1L on failure" semantic if you'd rather keep it strictly equivalent to the previous code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on PR #22454:
URL: https://github.com/apache/kafka/pull/22454#issuecomment-4658799062
thanks @chia7712 , I have updated with rebase.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3474179484
##########
core/src/main/scala/kafka/server/ControllerServer.scala:
##########
@@ -145,6 +145,11 @@ class ControllerServer(
if (linuxIoMetricsCollector.usable()) {
Review Comment:
Could you add a helper method to eliminate the duplicated code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22454:
URL: https://github.com/apache/kafka/pull/22454#issuecomment-4666563527
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3403672195
##########
core/src/main/scala/kafka/server/KafkaBroker.scala:
##########
@@ -120,5 +120,10 @@ trait KafkaBroker extends Logging {
if (linuxIoMetricsCollector.usable()) {
metricsGroup.newGauge("linux-disk-read-bytes", () => linuxIoMetricsCollector.readBytes())
Review Comment:
Thanks @chia7712 , I just pushed a refractor : cached fields are now volatile and the staleness check is consolidated into a single refreshIfStale() using double-checked locking. Each getter now does a volatile
read of lastUpdateMs plus the cached field with no monitor acquisition in the cached path, so a single JMX scrape across all 7 gauges takes the lock at most once per ms-window (and zero times in steady state). /proc/self/io is still read at most once per ms thanks to the existing cache. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3474170977
##########
server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java:
##########
@@ -35,36 +35,98 @@ public class LinuxIoMetricsCollector {
private static final Logger LOG = LoggerFactory.getLogger(LinuxIoMetricsCollector.class);
private static final String READ_BYTES_PREFIX = "read_bytes: ";
private static final String WRITE_BYTES_PREFIX = "write_bytes: ";
+ private static final String RCHAR_PREFIX = "rchar: ";
+ private static final String WCHAR_PREFIX = "wchar: ";
+ private static final String SYSCR_PREFIX = "syscr: ";
+ private static final String SYSCW_PREFIX = "syscw: ";
+ private static final String CANCELLED_WRITE_BYTES_PREFIX = "cancelled_write_bytes: ";
private final Time time;
private final Path path;
- private long lastUpdateMs = -1L;
- private long cachedReadBytes = 0L;
- private long cachedWriteBytes = 0L;
+ private volatile long lastUpdateMs = -1L;
+ private volatile long cachedReadBytes = 0L;
+ private volatile long cachedWriteBytes = 0L;
+ private volatile long cachedRchar = 0L;
+ private volatile long cachedWchar = 0L;
+ private volatile long cachedSyscr = 0L;
+ private volatile long cachedSyscw = 0L;
+ private volatile long cachedCancelledWriteBytes = 0L;
public LinuxIoMetricsCollector(String procRoot, Time time) {
this.time = time;
path = Paths.get(procRoot, "self", "io");
}
public long readBytes() {
- synchronized (this) {
- long curMs = time.milliseconds();
- if (curMs != lastUpdateMs) {
- updateValues(curMs);
- }
- return cachedReadBytes;
- }
+ refreshIfStale();
+ return cachedReadBytes;
}
public long writeBytes() {
- synchronized (this) {
- long curMs = time.milliseconds();
- if (curMs != lastUpdateMs) {
- updateValues(curMs);
+ refreshIfStale();
+ return cachedWriteBytes;
+ }
+
+ /**
+ * Returns the total number of characters read (includes cached reads).
+ * This value represents all read operations, including those satisfied by the page cache.
+ */
+ public long rchar() {
+ refreshIfStale();
+ return cachedRchar;
+ }
+
+ /**
+ * Returns the total number of characters written (includes cached writes).
+ * This value represents all write operations, including those that may not have reached disk.
+ */
+ public long wchar() {
+ refreshIfStale();
+ return cachedWchar;
+ }
+
+ /**
+ * Returns the number of read system calls.
+ * This metric helps identify I/O patterns and syscall overhead.
+ */
+ public long syscr() {
+ refreshIfStale();
+ return cachedSyscr;
+ }
+
+ /**
+ * Returns the number of write system calls.
+ * This metric helps identify I/O patterns and syscall overhead.
+ */
+ public long syscw() {
+ refreshIfStale();
+ return cachedSyscw;
+ }
+
+ /**
+ * Returns the number of bytes that were cancelled before being written.
+ * This can occur when a write is truncated or cancelled.
+ */
+ public long cancelledWriteBytes() {
+ refreshIfStale();
+ return cachedCancelledWriteBytes;
+ }
+
+ /**
+ * Refreshes all cached values from /proc/self/io if more than one millisecond has elapsed
+ * since the last refresh. Uses double-checked locking so getters return without acquiring
+ * the monitor in the common (cached) case — across all 7 gauges in a single JMX scrape
+ * the lock is taken at most once per ms-window, and zero times in steady state.
+ */
+ private void refreshIfStale() {
+ long curMs = time.milliseconds();
+ if (curMs != lastUpdateMs) {
+ synchronized (this) {
Review Comment:
How about using CAS instead of locking?
```java
private void refreshIfStale() {
long currentMs = time.milliseconds();
if (currentMs != lastUpdateMs) {
if (isUpdating.compareAndSet(false, true)) {
try {
currentMs = time.milliseconds(); // double check with the latest time
if (currentMs != lastUpdateMs) {
updateValues(currentMs);
}
} finally {
isUpdating.set(false);
}
}
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3485575900
##########
server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java:
##########
@@ -35,36 +35,98 @@ public class LinuxIoMetricsCollector {
private static final Logger LOG = LoggerFactory.getLogger(LinuxIoMetricsCollector.class);
private static final String READ_BYTES_PREFIX = "read_bytes: ";
private static final String WRITE_BYTES_PREFIX = "write_bytes: ";
+ private static final String RCHAR_PREFIX = "rchar: ";
+ private static final String WCHAR_PREFIX = "wchar: ";
+ private static final String SYSCR_PREFIX = "syscr: ";
+ private static final String SYSCW_PREFIX = "syscw: ";
+ private static final String CANCELLED_WRITE_BYTES_PREFIX = "cancelled_write_bytes: ";
private final Time time;
private final Path path;
- private long lastUpdateMs = -1L;
- private long cachedReadBytes = 0L;
- private long cachedWriteBytes = 0L;
+ private volatile long lastUpdateMs = -1L;
+ private volatile long cachedReadBytes = 0L;
+ private volatile long cachedWriteBytes = 0L;
+ private volatile long cachedRchar = 0L;
+ private volatile long cachedWchar = 0L;
+ private volatile long cachedSyscr = 0L;
+ private volatile long cachedSyscw = 0L;
+ private volatile long cachedCancelledWriteBytes = 0L;
public LinuxIoMetricsCollector(String procRoot, Time time) {
this.time = time;
path = Paths.get(procRoot, "self", "io");
}
public long readBytes() {
- synchronized (this) {
- long curMs = time.milliseconds();
- if (curMs != lastUpdateMs) {
- updateValues(curMs);
- }
- return cachedReadBytes;
- }
+ refreshIfStale();
+ return cachedReadBytes;
}
public long writeBytes() {
- synchronized (this) {
- long curMs = time.milliseconds();
- if (curMs != lastUpdateMs) {
- updateValues(curMs);
+ refreshIfStale();
+ return cachedWriteBytes;
+ }
+
+ /**
+ * Returns the total number of characters read (includes cached reads).
+ * This value represents all read operations, including those satisfied by the page cache.
+ */
+ public long rchar() {
+ refreshIfStale();
+ return cachedRchar;
+ }
+
+ /**
+ * Returns the total number of characters written (includes cached writes).
+ * This value represents all write operations, including those that may not have reached disk.
+ */
+ public long wchar() {
+ refreshIfStale();
+ return cachedWchar;
+ }
+
+ /**
+ * Returns the number of read system calls.
+ * This metric helps identify I/O patterns and syscall overhead.
+ */
+ public long syscr() {
+ refreshIfStale();
+ return cachedSyscr;
+ }
+
+ /**
+ * Returns the number of write system calls.
+ * This metric helps identify I/O patterns and syscall overhead.
+ */
+ public long syscw() {
+ refreshIfStale();
+ return cachedSyscw;
+ }
+
+ /**
+ * Returns the number of bytes that were cancelled before being written.
+ * This can occur when a write is truncated or cancelled.
+ */
+ public long cancelledWriteBytes() {
+ refreshIfStale();
+ return cachedCancelledWriteBytes;
+ }
+
+ /**
+ * Refreshes all cached values from /proc/self/io if more than one millisecond has elapsed
+ * since the last refresh. Uses double-checked locking so getters return without acquiring
+ * the monitor in the common (cached) case — across all 7 gauges in a single JMX scrape
+ * the lock is taken at most once per ms-window, and zero times in steady state.
+ */
+ private void refreshIfStale() {
+ long curMs = time.milliseconds();
+ if (curMs != lastUpdateMs) {
+ synchronized (this) {
Review Comment:
@chia7712 done. refreshIfStale now uses AtomicBoolean.compareAndSet; if another thread is mid-update, the caller falls through and reads the current cached value (safe since updateValues never publishes intermediate -1L). Also dropped synchronized from usable() - it's called once at construction time before any reader threads exist, so direct invocation is fine and the volatile writes carry happens-before to subsequent getters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #22454:
URL: https://github.com/apache/kafka/pull/22454#issuecomment-4646083437
@sahildevgon please rebase code to grab the fixes for CI :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #22454:
URL: https://github.com/apache/kafka/pull/22454
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3411315422
##########
server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java:
##########
@@ -79,32 +141,51 @@ public long writeBytes() {
* read_bytes: 0
* write_bytes: 0
* cancelled_write_bytes: 0
+ *
+ * <p>Caller must hold the monitor of this instance. {@code lastUpdateMs} is written last so
+ * that any thread observing the new timestamp via a volatile read also observes all updated
+ * cached fields.
*/
private boolean updateValues(long now) {
- synchronized (this) {
- try {
- cachedReadBytes = -1L;
- cachedWriteBytes = -1L;
- List<String> lines = Files.readAllLines(path, StandardCharsets.UTF_8);
- for (String line : lines) {
- if (line.startsWith(READ_BYTES_PREFIX)) {
- cachedReadBytes = Long.parseLong(line.substring(READ_BYTES_PREFIX.length()));
- } else if (line.startsWith(WRITE_BYTES_PREFIX)) {
- cachedWriteBytes = Long.parseLong(line.substring(WRITE_BYTES_PREFIX.length()));
- }
+ try {
+ cachedReadBytes = -1L;
Review Comment:
If you update them with temporary negative values, other concurrent calls taking the lock-free path will see this `-1L` state, right? It might be safer to use local variables for parsing and update the volatile fields all at once upon success
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on PR #22454:
URL: https://github.com/apache/kafka/pull/22454#issuecomment-4669227124
Thanks @m1a2st , I have added the same and PR for it is : https://github.com/apache/kafka/pull/22527. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on PR #22454:
URL: https://github.com/apache/kafka/pull/22454#issuecomment-4676774954
Thanks @FrankYang0529 , I have added the test to MetricsTest.scala. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #22454:
URL: https://github.com/apache/kafka/pull/22454#issuecomment-4668950853
FYI: https://github.com/apache/kafka/blob/trunk/docs/getting-started/upgrade.md
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20223: Extended I/O Metrics Collection [kafka]
Posted by "sahildevgon (via GitHub)" <gi...@apache.org>.
sahildevgon commented on code in PR #22454:
URL: https://github.com/apache/kafka/pull/22454#discussion_r3460198648
##########
server/src/main/java/org/apache/kafka/server/metrics/LinuxIoMetricsCollector.java:
##########
@@ -79,32 +141,61 @@ public long writeBytes() {
* read_bytes: 0
* write_bytes: 0
* cancelled_write_bytes: 0
+ *
+ * <p>Parses into local variables first, then publishes to the volatile cached fields only
+ * after parsing succeeds. This prevents lock-free readers from observing the transient
+ * -1L state during an update. {@code lastUpdateMs} is written last so any thread observing
+ * the new timestamp via a volatile read also observes all updated cached fields.
+ *
+ * <p>Caller must hold the monitor of this instance.
*/
private boolean updateValues(long now) {
- synchronized (this) {
- try {
- cachedReadBytes = -1L;
- cachedWriteBytes = -1L;
- List<String> lines = Files.readAllLines(path, StandardCharsets.UTF_8);
- for (String line : lines) {
- if (line.startsWith(READ_BYTES_PREFIX)) {
- cachedReadBytes = Long.parseLong(line.substring(READ_BYTES_PREFIX.length()));
- } else if (line.startsWith(WRITE_BYTES_PREFIX)) {
- cachedWriteBytes = Long.parseLong(line.substring(WRITE_BYTES_PREFIX.length()));
- }
+ long readBytes = -1L;
+ long writeBytes = -1L;
+ long rchar = -1L;
+ long wchar = -1L;
+ long syscr = -1L;
+ long syscw = -1L;
+ long cancelledWriteBytes = -1L;
+ try {
+ List<String> lines = Files.readAllLines(path, StandardCharsets.UTF_8);
+ for (String line : lines) {
+ if (line.startsWith(READ_BYTES_PREFIX)) {
+ readBytes = Long.parseLong(line.substring(READ_BYTES_PREFIX.length()));
+ } else if (line.startsWith(WRITE_BYTES_PREFIX)) {
+ writeBytes = Long.parseLong(line.substring(WRITE_BYTES_PREFIX.length()));
+ } else if (line.startsWith(RCHAR_PREFIX)) {
+ rchar = Long.parseLong(line.substring(RCHAR_PREFIX.length()));
+ } else if (line.startsWith(WCHAR_PREFIX)) {
+ wchar = Long.parseLong(line.substring(WCHAR_PREFIX.length()));
+ } else if (line.startsWith(SYSCR_PREFIX)) {
+ syscr = Long.parseLong(line.substring(SYSCR_PREFIX.length()));
+ } else if (line.startsWith(SYSCW_PREFIX)) {
+ syscw = Long.parseLong(line.substring(SYSCW_PREFIX.length()));
+ } else if (line.startsWith(CANCELLED_WRITE_BYTES_PREFIX)) {
+ cancelledWriteBytes = Long.parseLong(line.substring(CANCELLED_WRITE_BYTES_PREFIX.length()));
}
- lastUpdateMs = now;
- return true;
- } catch (Throwable t) {
- LOG.warn("Unable to update IO metrics", t);
- return false;
}
+ } catch (Throwable t) {
+ LOG.warn("Unable to update IO metrics", t);
Review Comment:
yeah, just updated the patch
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org