Add Kafka source observability metrics

[bagmarnikhil]

Description

Add metrics to the Kafka source, published through the existing Micrometer registry, so operators can diagnose scaling and reliability issues that were previously visible only in logs, or not at all:

• Scaling: per-partition lag, assigned partitions, active readers, and configured workers, so a bottleneck can be seen as partition-bound vs worker-bound.
• Rebalance activity: rebalance and partitions-revoked counts.
• Processing latency: poll to buffer handoff.
• Federated auth (azure_federated): token-refresh success and failure counts (failures tagged by a bounded errorType), and headroom in seconds to the next proactive token refresh.

The failure-handling counters from the issue (commit failures, buffer write failures, offset-reset failures) are intentionally left out here; they are covered by #7088.

Because the azure_federated callback handler is created reflectively by kafka-clients, its metrics object is handed off through a singleton (KafkaSourceAuthMetricsProvider), mirroring AwsCredentialsSupplierProvider, and is registered only for azure_federated sources.

Also make the azure_federated auth failure logging generic and actionable: log the underlying exception with its stack trace instead of asserting a single presumed cause, and drop the SENSITIVE marker from messages whose arguments are not credentials.

Issues Resolved

Resolves #7074

Check List

• New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
• New functionality has javadoc added
• Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check
here.

[github-actions]

✅ License Header Check Passed

All newly added files have proper license headers. Great work! 🎉

[github-actions]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit e570ebc.

⛔ Hard block: Issues at High severity or above will block this PR from merging.

Path	Line	Severity	Description
data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedTokenProvider.java	261	low	The SENSITIVE logging marker (DataPrepperMarkers.SENSITIVE) was removed from all three error-path log statements. While the refactored messages no longer include the STS role ARN (messages 1 and 2), the Azure token exchange error path (message 3) still logs the AADSTS error code without the SENSITIVE tag. If the SENSITIVE marker triggers downstream redaction or filtering in log aggregation pipelines, AADSTS diagnostic codes are now emitted without that protection. The net sensitivity is low (AADSTS codes are not credentials), but the behavioral change to the logging security boundary warrants review.

The table above displays the top 10 most important findings.

Total: 1 | Critical: 0 | High: 0 | Medium: 0 | Low: 1

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit e570ebc)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 Multiple PR themes Sub-PR theme: Add azure_federated auth token-refresh metrics and generalize failure logging Relevant files: data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedCallbackHandler.java data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedTokenProvider.java data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/KafkaSourceAuthMetrics.java data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/KafkaSourceAuthMetricsProvider.java data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedCallbackHandlerTest.java data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedTokenProviderTest.java data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/authenticator/KafkaSourceAuthMetricsProviderTest.java data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/authenticator/KafkaSourceAuthMetricsTest.java Sub-PR theme: Add Kafka consumer scaling, rebalance, and processing-latency metrics Relevant files: data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicConsumerMetrics.java data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumerTest.java data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicConsumerMetricsLagGaugeTest.java data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicConsumerMetricsTests.java

⚡ Recommended focus areas for review Incorrect refresh deadline argument authMetrics.recordRefresh(token.lifetimeMs() - SKEW_BUFFER_MS) passes a duration (lifetime minus skew) rather than an absolute epoch-millis deadline. However, KafkaSourceAuthMetrics.recordRefresh documents its argument as "absolute epoch-millis instant" and computes headroom as (deadline - clock.getAsLong()) / 1000.0. Passing a duration (e.g., ~3599000) will make headroom immediately negative and floored to 0, so timeToTokenRefresh will always read 0 in production. The unit test in AzureFederatedTokenProviderTest.getToken_onSuccess_incrementsRefreshCountAndSetsExpiryHeadroom only asserts greaterThan(0.0) but that relies on AzureFederatedOAuthBearerToken.lifetimeMs() returning something greater than SKEW_BUFFER_MS + now which is unlikely for typical values. if (authMetrics != null) { // Headroom to the proactive re-mint deadline (expiry minus skew), not raw expiry. authMetrics.recordRefresh(token.lifetimeMs() - SKEW_BUFFER_MS); } Unbounded per-partition gauge growth registeredLagGauges is only ever added to; when partitions are revoked (clearPartitionLag), the gauge remains registered and perPartitionLag retains the entry forever. Over long-lived consumers with frequent rebalances and shifting partition assignments across a topic, this leaks meter entries and map entries. Consider removing the partition from registeredLagGauges and perPartitionLag on revocation, or documenting the intentional retention. public void clearPartitionLag(final int partition) { perPartitionLag.put(String.valueOf(partition), 0.0); } Auth metrics prefix shared across topics KafkaSourceAuthMetricsProvider.set is called once per source with a single KafkaSourceAuthMetrics, but the source can have multiple topics each constructing its own KafkaTopicConsumerMetrics. Token refresh metrics won't be scoped per-topic and, per the singleton's own doc, if multiple federated Kafka sources exist in the same JVM the second set() silently overwrites the first. Confirm this last-writer-wins behavior is acceptable in shared-JVM deployments and consider logging a warning on overwrite. if (usesAzureFederatedAuth()) { KafkaSourceAuthMetricsProvider.getInstance().set(new KafkaSourceAuthMetrics(pluginMetrics)); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to e570ebc

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Pass absolute deadline, not duration, to gauge The recordRefresh contract expects an absolute epoch-millis deadline (used to compute headroom via deadline - clock.getAsLong()), but token.lifetimeMs() - SKEW_BUFFER_MS is a duration, not an absolute timestamp. Pass the absolute deadline instead (e.g., token.startTimeMs() + token.lifetimeMs() - SKEW_BUFFER_MS), otherwise the gauge will compute a huge negative headroom clamped to 0. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedTokenProvider.java [153-157] cached = new CachedToken(token); if (authMetrics != null) { // Headroom to the proactive re-mint deadline (expiry minus skew), not raw expiry. - authMetrics.recordRefresh(token.lifetimeMs() - SKEW_BUFFER_MS); + authMetrics.recordRefresh(token.startTimeMs() + token.lifetimeMs() - SKEW_BUFFER_MS); } Suggestion importance[1-10]: 9 __ Why: This is a correct and significant bug catch: recordRefresh expects an absolute epoch-millis deadline (subtracted from clock.getAsLong() in currentRefreshHeadroomSeconds), but passing token.lifetimeMs() - SKEW_BUFFER_MS provides only a duration, causing the gauge to always floor at 0 in production (though tests using controllable clocks starting at epoch 0 may pass).	High
General	Avoid overwriting auth metrics singleton on restart KafkaSource.start() may be invoked more than once in a JVM (e.g., pipeline restart or multiple sources). Each call replaces the singleton with a new KafkaSourceAuthMetrics bound to a fresh PluginMetrics, re-registering the gauge which the registry deduplicates but the previous instance's refreshDeadlineMs state is orphaned. Consider setting only once, or explicitly clearing/unregistering on stop to avoid stale gauges lingering after pipeline shutdown. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/source/KafkaSource.java [137-139] -if (usesAzureFederatedAuth()) { +if (usesAzureFederatedAuth() && KafkaSourceAuthMetricsProvider.getInstance().getAuthMetrics() == null) { KafkaSourceAuthMetricsProvider.getInstance().set(new KafkaSourceAuthMetrics(pluginMetrics)); } Suggestion importance[1-10]: 5 __ Why: Legitimate concern about restart/multi-source scenarios and stale gauges, though the PR documentation acknowledges the last-writer-wins limitation. The suggested fix is reasonable but a partial mitigation.	Low
	Record latency in a finally block The processing-latency timer only records on the success path; if iterateRecordPartitions or acknowledgementSet.complete() throws, no latency is recorded and, worse, exceptions from the batch path are not routed through the existing AuthenticationException catch (they will bubble up unmeasured). Consider using a try/finally to record latency for every non-empty batch regardless of outcome so the timer accurately reflects processing time including failures. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java [237-239] if (Objects.nonNull(records) && !records.isEmpty() && records.count() > 0) { // Time poll() return through buffer handoff, once per non-empty batch. final long processingStartMillis = System.currentTimeMillis(); + try { + Map<TopicPartition, CommitOffsetRange> offsets = new HashMap<>(); + // ... existing batch body ... + } finally { + topicMetrics.recordProcessingLatency(System.currentTimeMillis() - processingStartMillis); + } Suggestion importance[1-10]: 4 __ Why: Valid observation that failed batches won't record latency, but this may be intentional (recording only successful processing time). The improvement is minor.	Low
	Guard against null per-consumer metrics map foldAssignedPartitions calls consumerMetrics.getOrDefault(ASSIGNED_PARTITIONS, 0.0) before any update() populates the map. During register(), the map is empty and this works. However, after the write-time inversion was removed, if update() has never been called for a newly registered consumer, consumerMetrics may still be empty which is fine—but the synchronized block still calls getOrDefault on the shared map. Consider validating that consumerMetrics is never null (it is created in register()), otherwise foldAssignedPartitions will NPE. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicConsumerMetrics.java [112-115] +pluginMetrics.gauge(getTopicMetricName(NUMBER_OF_ASSIGNED_PARTITIONS, topicNameInMetrics), metricValues, + mv -> foldAssignedPartitions(mv, raw -> raw)); +pluginMetrics.gauge(getTopicMetricName(NUMBER_OF_ACTIVE_READERS, topicNameInMetrics), metricValues, + mv -> foldAssignedPartitions(mv, raw -> raw > 0 ? 1.0 : 0.0)); - Suggestion importance[1-10]: 2 __ Why: The suggestion is speculative; register() always populates the map before any update(), and the improved code is identical to the existing code, providing no real change.	Low

---

Previous suggestions

Suggestions up to commit c6e3766

Category	Suggestion	Impact
Possible issue	Pass absolute deadline, not a duration recordRefresh expects an absolute epoch-millis deadline (the gauge computes deadline - clock.getAsLong()), but token.lifetimeMs() - SKEW_BUFFER_MS is a duration, not an absolute timestamp. This will cause the headroom gauge to report a huge negative number floored to 0. Pass an absolute deadline based on the token's start time. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedTokenProvider.java [155-161] if (authMetrics != null) { - // Report headroom to the REFRESH deadline (lifetime minus the skew buffer), not to the - // token's raw expiry: the provider proactively re-mints SKEW_BUFFER_MS early, so healthy - // operation should show the gauge counting down toward 0 at the refresh point, letting a - // "headroom < N" alarm be set meaningfully rather than flooring at the skew buffer. - authMetrics.recordRefresh(token.lifetimeMs() - SKEW_BUFFER_MS); + authMetrics.recordRefresh(token.startTimeMs() + token.lifetimeMs() - SKEW_BUFFER_MS); } Suggestion importance[1-10]: 9 __ Why: This is a critical bug: recordRefresh treats its argument as an absolute epoch-millis deadline (gauge computes deadline - clock.getAsLong()), but the code passes token.lifetimeMs() - SKEW_BUFFER_MS which is a duration. This would cause the headroom gauge to always report 0, defeating the purpose of the new metric.	High
General	Ensure consumer is always closed If topicMetrics.deregister(consumer) throws for any reason, consumer.close() will be skipped, leaking the Kafka consumer and its network/heartbeat threads. Wrap the deregister call so that consumer close is always attempted. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java [627-630] public void closeConsumer(){ - topicMetrics.deregister(consumer); - consumer.close(); + try { + topicMetrics.deregister(consumer); + } finally { + consumer.close(); + } } Suggestion importance[1-10]: 6 __ Why: Reasonable defensive change: wrapping deregister in try/finally ensures consumer.close() runs even if metric deregistration throws, preventing potential resource leaks.	Low
	Guard cast of Kafka metric value value.metricValue() from kafka-clients can return non-Double types (e.g., a String "NaN" placeholder or other object), and an unconditional cast to Double will throw ClassCastException and break the whole update loop. Guard the cast by checking instanceof Double before casting. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicConsumerMetrics.java [322-335] if (metricName.equals("records-lag-max")) { final String partition = metric.tags().get("partition"); - final double lag = (Double) value.metricValue(); - if (!Double.isNaN(lag) && !Double.isInfinite(lag)) { - // plain put (latest value), NOT merge(Math::max): a monotonic max would turn - // this into a high-water-mark rather than a live lag gauge. - perPartitionLag.put(partition, lag); - if (registeredLagGauges.add(partition)) { + final Object raw = value.metricValue(); + if (raw instanceof Double) { + final double lag = (Double) raw; + if (!Double.isNaN(lag) && !Double.isInfinite(lag)) { + perPartitionLag.put(partition, lag); + if (registeredLagGauges.add(partition)) { Suggestion importance[1-10]: 5 __ Why: Defensive cast guard is reasonable since metricValue() returns Object, but Kafka's records-lag-max is documented as a Double gauge, so a ClassCastException is unlikely in practice. Still improves robustness.	Low
	Document/enforce gauge weak-ref retention Micrometer's DefaultGauge holds a WEAK reference to this, so if the caller (e.g. KafkaSource.start) does not retain the created KafkaSourceAuthMetrics instance, the gauge will collect it and permanently report NaN. Only KafkaSourceAuthMetricsProvider retains it via a volatile field, so a subsequent set(null) (as done in the test teardown) also drops the strong reference at runtime. Consider retaining a strong self-reference from a static holder tied to lifetime, or documenting/enforcing the retention contract in KafkaSource. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/KafkaSourceAuthMetrics.java [63] pluginMetrics.gauge(TIME_TO_TOKEN_REFRESH, this, KafkaSourceAuthMetrics::currentRefreshHeadroomSeconds); +// Retention contract: KafkaSourceAuthMetricsProvider must hold this instance strongly for the +// source's lifetime, otherwise the gauge's weak reference will be GC'd and value becomes NaN. Suggestion importance[1-10]: 3 __ Why: The retention contract is already documented in the existing comment above the gauge registration. The suggestion mainly restates this, offering only marginal improvement.	Low

Suggestions up to commit 249e04a

Category	Suggestion	Impact
Possible issue	Pass absolute deadline to recordRefresh recordRefresh expects an absolute epoch-millis deadline (the gauge computes deadline - clock.getAsLong()), but here you pass token.lifetimeMs() - SKEW_BUFFER_MS, which is a duration, not an absolute timestamp. This will cause the headroom gauge to report a huge negative value (floored to 0) or otherwise incorrect values. Pass an absolute deadline based on the token's start time plus lifetime minus skew. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedTokenProvider.java [155-161] cached = new CachedToken(token); if (authMetrics != null) { - // Report headroom to the REFRESH deadline (lifetime minus the skew buffer), not to the - // token's raw expiry: the provider proactively re-mints SKEW_BUFFER_MS early, so healthy - // operation should show the gauge counting down toward 0 at the refresh point, letting a - // "headroom < N" alarm be set meaningfully rather than flooring at the skew buffer. - authMetrics.recordRefresh(token.lifetimeMs() - SKEW_BUFFER_MS); + authMetrics.recordRefresh(token.startTimeMs() + token.lifetimeMs() - SKEW_BUFFER_MS); } Suggestion importance[1-10]: 9 __ Why: This appears to be a correct catch of a significant bug: recordRefresh expects an absolute epoch-millis deadline (per its Javadoc and the gauge computation deadline - clock.getAsLong()), but the code passes a duration (token.lifetimeMs() - SKEW_BUFFER_MS), which would make the headroom gauge report incorrect (floored-to-zero) values.	High
General	Avoid cross-consumer lag overwrites With multiple consumers in the same JVM feeding this shared KafkaTopicConsumerMetrics instance, each consumer's update() will put its own lag for the same partition, so the last writer wins and lag from other consumers is overwritten. Additionally, clearPartitionLag from a revoke on one consumer can wipe a value another consumer just wrote. Consider keying perPartitionLag by (consumer, partition) or otherwise scoping updates to the owning consumer. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicConsumerMetrics.java [322-336] if (metricName.equals("records-lag-max")) { final String partition = metric.tags().get("partition"); final double lag = (Double) value.metricValue(); if (!Double.isNaN(lag) && !Double.isInfinite(lag)) { - // plain put (latest value), NOT merge(Math::max): a monotonic max would turn - // this into a high-water-mark rather than a live lag gauge. + // Consider scoping the key by consumer identity to avoid cross-consumer overwrites. perPartitionLag.put(partition, lag); if (registeredLagGauges.add(partition)) { - // topic pinned in the name regardless of topicNameInMetrics: partition tags collide across topics otherwise. pluginMetrics.gaugeWithTags(getTopicMetricName(RECORDS_LAG_PER_PARTITION, true), List.of(Tag.of("partition", partition)), perPartitionLag, m -> m.getOrDefault(partition, 0.0)); } } } continue; Suggestion importance[1-10]: 6 __ Why: Valid concern: multiple consumers sharing a KafkaTopicConsumerMetrics instance will overwrite each other's per-partition lag since Kafka only assigns a partition to one consumer at a time within a group, but transient rebalance states could cause cross-consumer interference with clearPartitionLag.	Low
	Avoid re-registering auth metrics on restart KafkaSourceAuthMetricsProvider is a process-global singleton and start() may be invoked multiple times (or by multiple sources); each call replaces the previous KafkaSourceAuthMetrics and re-registers gauges/counters against pluginMetrics, which can duplicate meters or orphan the previously-registered pull-model gauge (Micrometer's gauge holds a weak reference to the passed object). Set it once and reuse, or ensure the previous instance is retained for the source's lifetime. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/source/KafkaSource.java [139-141] - if (usesAzureFederatedAuth()) { + if (usesAzureFederatedAuth() && KafkaSourceAuthMetricsProvider.getInstance().getAuthMetrics() == null) { KafkaSourceAuthMetricsProvider.getInstance().set(new KafkaSourceAuthMetrics(pluginMetrics)); } Suggestion importance[1-10]: 5 __ Why: Reasonable concern about repeated start() invocations replacing the singleton and potentially orphaning the weakly-referenced gauge; however the target deployment is one source per JVM as documented, limiting real-world impact.	Low
	Record latency even on failure paths recordProcessingLatency is only invoked on the success path; if iterateRecordPartitions/processRecords throws (e.g., buffer wedged, deserialization error), the timer never records that batch, biasing the metric toward healthy batches only. Move the record call into a finally block so backpressure/failure latency is also captured, which is the stated intent of the metric. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java [254-257] - } else { - acknowledgementSet.complete(); - numberOfAcksPending.incrementAndGet(); + try { + ... + if (!acknowledgementsEnabled) { + offsets.forEach((partition, offsetRange) -> { + updateOffsetsToCommit(partition, new OffsetAndMetadata(offsetRange.getOffsets().getMaximum() + 1)); + numRecordsCommitted += offsetRange.getOffsets().getMaximum() - offsetRange.getOffsets().getMinimum() + 1; + }); + } else { + acknowledgementSet.complete(); + numberOfAcksPending.incrementAndGet(); + } + } finally { + topicMetrics.recordProcessingLatency(System.currentTimeMillis() - processingStartMillis); } - topicMetrics.recordProcessingLatency(System.currentTimeMillis() - processingStartMillis); - } Suggestion importance[1-10]: 4 __ Why: Valid observation that the timer only records success paths, but the code comment explicitly states the intent is to capture "in-consumer processing + backpressure", and exceptions typically indicate a distinct failure mode tracked by other metrics. Impact is moderate.	Low

[github-actions]

Persistent review updated to latest commit c6e3766

[github-actions]

Persistent review updated to latest commit e570ebc

[bagmarnikhil]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit e570ebc.

⛔ Hard block: Issues at High severity or above will block this PR from merging.
Path Line Severity Description
data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/AzureFederatedTokenProvider.java 261 low The SENSITIVE logging marker (DataPrepperMarkers.SENSITIVE) was removed from all three error-path log statements. While the refactored messages no longer include the STS role ARN (messages 1 and 2), the Azure token exchange error path (message 3) still logs the AADSTS error code without the SENSITIVE tag. If the SENSITIVE marker triggers downstream redaction or filtering in log aggregation pipelines, AADSTS diagnostic codes are now emitted without that protection. The net sensitivity is low (AADSTS codes are not credentials), but the behavioral change to the logging security boundary warrants review.

The table above displays the top 10 most important findings. Total: 1 | Critical: 0 | High: 0 | Medium: 0 | Low: 1

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

Intentional. The SENSITIVE marker only masks {} args, and none of these three lines pass a credential — they log an STS/HTTP status and the AADSTS error code (e.g. AADSTS700016), which is a public diagnostic identifier, not a secret. The token/assertion is never logged. Removing the marker makes these operator-facing errors readable instead of ******. Low severity, no credential exposure — leaving as-is.

[bagmarnikhil]

PR Reviewer Guide 🔍

(Review updated until commit e570ebc)

Here are some key observations to aid the review process:
🧪 PR contains tests
🔒 No security concerns identified
✅ No TODO sections
🔀 Multiple PR themes

Sub-PR theme: Add azure_federated auth token-refresh metrics and generalize failure logging

Sub-PR theme: Add Kafka consumer scaling, rebalance, and processing-latency metrics

⚡ Recommended focus areas for review

Incorrect refresh deadline argument

authMetrics.recordRefresh(token.lifetimeMs() - SKEW_BUFFER_MS) passes a duration (lifetime minus skew) rather than an absolute epoch-millis deadline. However, KafkaSourceAuthMetrics.recordRefresh documents its argument as "absolute epoch-millis instant" and computes headroom as (deadline - clock.getAsLong()) / 1000.0. Passing a duration (e.g., ~3599000) will make headroom immediately negative and floored to 0, so timeToTokenRefresh will always read 0 in production. The unit test in AzureFederatedTokenProviderTest.getToken_onSuccess_incrementsRefreshCountAndSetsExpiryHeadroom only asserts greaterThan(0.0) but that relies on AzureFederatedOAuthBearerToken.lifetimeMs() returning something greater than SKEW_BUFFER_MS + now which is unlikely for typical values.

Unbounded per-partition gauge growth

registeredLagGauges is only ever added to; when partitions are revoked (clearPartitionLag), the gauge remains registered and perPartitionLag retains the entry forever. Over long-lived consumers with frequent rebalances and shifting partition assignments across a topic, this leaks meter entries and map entries. Consider removing the partition from registeredLagGauges and perPartitionLag on revocation, or documenting the intentional retention.

Auth metrics prefix shared across topics

KafkaSourceAuthMetricsProvider.set is called once per source with a single KafkaSourceAuthMetrics, but the source can have multiple topics each constructing its own KafkaTopicConsumerMetrics. Token refresh metrics won't be scoped per-topic and, per the singleton's own doc, if multiple federated Kafka sources exist in the same JVM the second set() silently overwrites the first. Confirm this last-writer-wins behavior is acceptable in shared-JVM deployments and consider logging a warning on overwrite.

---

1. Refresh deadline — not a bug. token.lifetimeMs() is absolute epoch-millis (currentTimeMillis() + expiresIn*1000, the Kafka OAuthBearerToken contract), so
lifetimeMs() - SKEW_BUFFER_MS is the absolute deadline recordRefresh expects. Tests use an injected clock and assert the countdown.

2. Per-partition gauges — bounded, intentional. Cardinality is capped by partition count; clearPartitionLag zeroes the value and reassignment refreshes it. True
gauge removal belongs with the rebalance handling in a separate PR.

3. Auth metrics are source-scoped by design. Token refresh is per-source, not per-topic, so one object is correct. Last-writer-wins across two federated sources in
one JVM is the documented limitation inherited from AwsCredentialsSupplierProvider; target is one source per container.