Add metrics for Kafka offset commit failures#38613
Conversation
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request enhances the monitoring capabilities of the Kafka IO connector by adding specific metrics to track offset commit failures. By integrating Beam's counter metrics, it provides better visibility into the health of Kafka offset commits, facilitating easier debugging and alerting for production pipelines. Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces metrics to track Kafka offset commit failures within the KafkaCommitOffset class. The feedback suggests removing the retriesExhausted metric and its associated increments, as no retry logic is currently implemented, making the metric redundant. Additionally, the reviewer recommends removing the excessive vertical whitespace introduced throughout the file to maintain consistency with the project's coding style and improve readability.
| private final Counter commitFailures = | ||
| Metrics.counter(CommitOffsetDoFn.class, "commit-failures"); | ||
|
|
||
| private final Counter retriesExhausted = | ||
| Metrics.counter(CommitOffsetDoFn.class, "retries-exhausted"); | ||
|
|
There was a problem hiding this comment.
The retriesExhausted metric is currently redundant with commitFailures because no retry logic is implemented yet. It is better to introduce this metric when retries are actually added. Additionally, there is an excessive amount of vertical whitespace being introduced throughout the file which deviates from the existing code style.
| private final Counter commitFailures = | |
| Metrics.counter(CommitOffsetDoFn.class, "commit-failures"); | |
| private final Counter retriesExhausted = | |
| Metrics.counter(CommitOffsetDoFn.class, "retries-exhausted"); | |
| private final Counter commitFailures = | |
| Metrics.counter(CommitOffsetDoFn.class, "commit-failures"); | |
|
|
||
| Map<String, Object> updatedConsumerConfig = | ||
| overrideBootstrapServersConfig(consumerConfig, element.getKey()); | ||
|
|
||
| try (Consumer<byte[], byte[]> consumer = consumerFactoryFn.apply(updatedConsumerConfig)) { | ||
|
|
||
| try { | ||
| consumer.commitSync( | ||
| Collections.singletonMap( | ||
| element.getKey().getTopicPartition(), | ||
| new OffsetAndMetadata(element.getValue() + 1))); | ||
|
|
||
| } catch (Exception e) { | ||
| // TODO: consider retrying. | ||
|
|
||
| commitFailures.inc(); | ||
| retriesExhausted.inc(); | ||
|
|
||
| // TODO: consider retrying and increment retry-attempt metrics. | ||
| LOG.warn("Getting exception when committing offset: {}", e.getMessage()); | ||
| } | ||
| } |
There was a problem hiding this comment.
Removing the redundant metric increment and cleaning up the excessive vertical whitespace to improve code density and readability.
Map<String, Object> updatedConsumerConfig =
overrideBootstrapServersConfig(consumerConfig, element.getKey());
try (Consumer<byte[], byte[]> consumer = consumerFactoryFn.apply(updatedConsumerConfig)) {
try {
consumer.commitSync(
Collections.singletonMap(
element.getKey().getTopicPartition(),
new OffsetAndMetadata(element.getValue() + 1)));
} catch (Exception e) {
commitFailures.inc();
// TODO: consider retrying and increment retry-attempt metrics.
LOG.warn("Getting exception when committing offset: {}", e.getMessage());
}
}|
|
||
| checkState( | ||
| currentConfig.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG) | ||
| || description.getBootStrapServers() != null); | ||
|
|
||
| Map<String, Object> config = new HashMap<>(currentConfig); | ||
|
|
||
| if (description.getBootStrapServers() != null | ||
| && !description.getBootStrapServers().isEmpty()) { | ||
|
|
||
| config.put( | ||
| ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, | ||
| String.join(",", description.getBootStrapServers())); | ||
| } | ||
|
|
There was a problem hiding this comment.
Removing unnecessary blank lines to maintain consistency with the project's coding style.
checkState(
currentConfig.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)
|| description.getBootStrapServers() != null);
Map<String, Object> config = new HashMap<>(currentConfig);
if (description.getBootStrapServers() != null
&& !description.getBootStrapServers().isEmpty()) {
config.put(
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
String.join(",", description.getBootStrapServers()));
}|
|
||
| try { | ||
|
|
||
| PCollection<KV<KafkaSourceDescriptor, Long>> offsets; | ||
|
|
||
| if (use259implementation) { | ||
|
|
||
| offsets = | ||
| input.apply( | ||
| MapElements.into(new TypeDescriptor<KV<KafkaSourceDescriptor, Long>>() {}) | ||
| .via(element -> KV.of(element.getKey(), element.getValue().getOffset()))); | ||
|
|
||
| } else { | ||
|
|
||
| // Reduce the amount of data to combine by calculating a max within the generally dense | ||
| // bundles of reading | ||
| // from a Kafka partition. | ||
| offsets = input.apply(ParDo.of(new MaxOffsetFn<>())); | ||
| } | ||
|
|
There was a problem hiding this comment.
Cleaning up excessive whitespace in the expand method to improve readability.
try {
PCollection<KV<KafkaSourceDescriptor, Long>> offsets;
if (use259implementation) {
offsets =
input.apply(
MapElements.into(new TypeDescriptor<KV<KafkaSourceDescriptor, Long>>() {})
.via(element -> KV.of(element.getKey(), element.getValue().getOffset())));
} else {
// Reduce the amount of data to combine by calculating a max within the generally dense
// bundles of reading
// from a Kafka partition.
offsets = input.apply(ParDo.of(new MaxOffsetFn<>()));
}
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #38613 +/- ##
=========================================
Coverage 58.07% 58.07%
- Complexity 12941 12945 +4
=========================================
Files 2511 2511
Lines 262223 262265 +42
Branches 10652 10655 +3
=========================================
+ Hits 152282 152308 +26
- Misses 104243 104258 +15
- Partials 5698 5699 +1
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
|
Assigning reviewers: R: @Abacn for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
1 participant
Summary
Added Beam metrics in
KafkaCommitOffset.CommitOffsetDoFnto improve visibility into offset commit failures.Added metrics
commit-failuresretries-exhaustedThese counters help monitor failed offset commits through existing Beam monitoring dashboards.
Retry-attempt metrics can be added once retry logic is implemented.