Add failure metrics to the Kafka source by harshwadhawe · Pull Request #7088 · opensearch-project/data-prepper

harshwadhawe · 2026-08-12T17:30:30Z

Description

Three failure paths in the Kafka source are currently invisible to operators. They are written to the log and never counted:

Offset commit failures (KafkaCustomConsumer#commitOffsets): both the RebalanceInProgressException branch and the generic branch.
Offset reset failures (KafkaCustomConsumer#resetOffsets): the seek back to the last committed offset after a negative acknowledgement.
Buffer write failures: the non-SizeOverflowException branch of the buffer write retry loop. This one is only logged at debug, so it is invisible even at default log levels. Size overflows are already counted via numberOfBufferSizeOverflows, but write timeouts are not.

These are the failures that lead to reprocessed records and stalled pipelines, so today an operator cannot distinguish "the pipeline is slow" from "commits are failing" without turning on debug logging.

This adds one counter for each, following the existing constant/field/initializer/getter pattern in KafkaTopicConsumerMetrics:

numberOfCommitFailures
numberOfOffsetResetFailures
numberOfBufferWriteFailures

The change to KafkaCustomConsumer is five increment() calls; no control flow is altered.

Scope. This is the failure-handling portion of #7074 only. The other areas in that issue the assigned-partition gauge, rebalance revoked/lost counters, poll-to-buffer latency, per-partition lag, and federated auth health are intentionally left out. Per-partition lag multiplies series by topics × partitions and needs an opt-in decision, and the federated auth callback handler is instantiated by Kafka from the JAAS configuration with no PluginMetrics available, so it needs a design discussion first. Happy to follow up with those in separate PRs once there is agreement on the issue.

Note on the tests. The first test run failed two pre-existing tests, which turned out to be a useful signal: topicMetrics is a mock in KafkaCustomConsumerTest, so an unstubbed new getter returned null and the resulting NPE fired inside the catch block, preventing the recovery path from running. In production topicMetrics is always a real instance, so this was test wiring, fixed by stubbing the three getters in setUp alongside the existing counters. It does confirm the increments sit on paths that actually execute.

Issues Resolved

Partially addresses #7074. Deliberately not using "Resolves", since this covers one of the five areas requested in that issue.

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

On documentation: the Kafka source currently has no metrics documentation at all. There is no ## Metrics section on the kafka source page (13 of the 19 source pages have one) and none in the plugin README, so none of the existing counters are documented either.

I would rather fix that in one pass than document only the three new ones. Need some guidance for updating the documentation.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Three failures in the Kafka source were only written to logs: failed offset commits, failed offset resets after a negative acknowledgement, and failed buffer writes. Operators had no way to see them in dashboards or alerts. This adds a counter for each one. Partially addresses opensearch-project#7074. Signed-off-by: Harsh Wadhawe <harsh.wadhawe@tamu.edu>

Signed-off-by: Harsh Wadhawe <harsh.wadhawe@tamu.edu>

github-actions · 2026-08-12T17:31:38Z

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests
🔒 No security concerns identified
✅ No TODO sections
🔀 No multiple PR themes
⚡ No major issues detected

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 opensearch-project#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. Resolves opensearch-project#7074 Signed-off-by: Nikhil Bagmar <nikhilbagmar73@gmail.com>

bagmarnikhil

Thanks for picking this up, and for keeping the scope small.

These are a subset of #7074. I have a separate PR coming for the rest (partition assignment, rebalance counts, per-partition lag, processing latency, federated-auth
health).

Left one non-blocking naming note inline on numberOfOffsetResetFailures (numberOfSeekFailures reads clearer). Otherwise LGTM.

bagmarnikhil · 2026-08-17T23:32:01Z

+        return numberOfCommitFailures;
+    }
+
+    public Counter getNumberOfOffsetResetFailures() {


The name numberOfOffsetResetFailures is easy to misread. In Kafka, "offset reset" usually points at the auto.offset.reset policy (earliest/latest when there is no committed offset). This counter is about a different thing: the seek back to the last committed offset in resetOffsets() after a negative acknowledgement, so those records get redelivered instead of skipped.

Two details about what it actually counts, in case the name should reflect them:

It wraps both the consumer.committed(partition) lookup and the seek, so a failure in either step increments it, not just the seek.

It increments per partition inside the loop, once per failed reset attempt.

Because it fires only on the negative-ack redelivery path, a name tied to that path reads more clearly on a dashboard and avoids the auto.offset.reset confusion. A couple of options:

numberOfSeekFailures — matches Kafka's own API term (consumer.seek) and stays short.

numberOfNegativeAckSeekFailures — longer, but explicit about the trigger.

I like numberOfSeekFailures.

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 opensearch-project#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. Resolves opensearch-project#7074 Signed-off-by: Nikhil Bagmar <nikhilbagmar73@gmail.com>

dlvenable

Thank you @harshwadhawe for this contribution!

dlvenable · 2026-08-18T21:23:37Z

+        return numberOfCommitFailures;
+    }
+
+    public Counter getNumberOfOffsetResetFailures() {


I like numberOfSeekFailures.

dlvenable · 2026-08-18T21:29:37Z

                    final long epoch = getCurrentTimeNanos();
                    ownedPartitionsEpoch.put(partition, epoch);
                } catch (Exception e) {
+                    topicMetrics.getNumberOfOffsetResetFailures().increment();


This will catch both a seek failure and a commit error. Should these be the same metric?

harshwadhawe added 2 commits August 12, 2026 12:16

Use reflection in test instead of widening method visibility

3d8e3ff

Signed-off-by: Harsh Wadhawe <harsh.wadhawe@tamu.edu>

harshwadhawe requested review from KarstenSchnitter, Zhangxunmt, dinujoh, divbok, dlvenable, graytaylor0, kkondaka, oeyh, san81, sb2k16, srikanthjg and srikanthpadakanti as code owners August 12, 2026 17:30

harshwadhawe mentioned this pull request Aug 12, 2026

Add observability metrics to the Kafka source #7074

Open

bagmarnikhil mentioned this pull request Aug 17, 2026

Add Kafka source observability metrics #7102

Open

4 tasks

bagmarnikhil reviewed Aug 18, 2026

View reviewed changes

dlvenable requested changes Aug 18, 2026

View reviewed changes

Navigation Menu

Uh oh!

Add failure metrics to the Kafka source - #7088

Add failure metrics to the Kafka source#7088
harshwadhawe wants to merge 2 commits into
opensearch-project:mainfrom
harshwadhawe:feature/kafka-source-failure-metrics

harshwadhawe commented Aug 12, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Aug 12, 2026

Uh oh!

bagmarnikhil left a comment

Uh oh!

bagmarnikhil Aug 17, 2026

Uh oh!

dlvenable Aug 18, 2026

Uh oh!

dlvenable left a comment

Uh oh!

dlvenable Aug 18, 2026

Uh oh!

dlvenable Aug 18, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

harshwadhawe commented Aug 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Issues Resolved

Check List

Uh oh!

github-actions Bot commented Aug 12, 2026

PR Reviewer Guide 🔍

Uh oh!

bagmarnikhil left a comment

Choose a reason for hiding this comment

Uh oh!

bagmarnikhil Aug 17, 2026

Choose a reason for hiding this comment

Uh oh!

dlvenable Aug 18, 2026

Choose a reason for hiding this comment

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

dlvenable Aug 18, 2026

Choose a reason for hiding this comment

Uh oh!

dlvenable Aug 18, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

harshwadhawe commented Aug 12, 2026 •

edited

Loading