Bug/1097 consumer offset commit #1183
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Contributor
📝 WalkthroughWalkthroughReworks Kafka consumer loop to reset per-iteration state, centralize offset commits, add explicit retry/error handlers, and commit offsets only after successful processing or successful retry. Adds ProduceAsync result checks in multiple consumers and new tests exercising producer-failure retry paths (one test duplicated). Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
Author
|
@coderabbitai review |
Contributor
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs (1)
97-113: Critical: Retry failure terminates the entire consumer.The
throwat line 106 will propagate out of the catch block and exit theConsumeMessagemethod, terminating the while loop and stopping the entire consumer. This means a single retry failure permanently stops message processing for this consumer until manually restarted.While preventing offset commits on retry failure is correct, terminating the consumer has severe operational consequences:
- All subsequent messages in the topic remain unprocessed
- Requires manual intervention to restart the consumer
- Could cause system-wide impact for critical consumers
Consider this alternative approach that prevents offset commit without terminating the consumer:
catch (Exception ex) { - bool retrySucceeded = false; - try - { - await retryMessageFunc(message!); - retrySucceeded = true; - } - catch (Exception retryEx) - { - _logger.LogError(retryEx, "// {Class} // ConsumeMessage // An error occurred while retrying message processing", GetType().Name); - throw; // prevent offset commit - } - - if (retrySucceeded && consumeResult != null) + try + { + await retryMessageFunc(message!); + // Retry succeeded - safe to commit offset + if (consumeResult != null) + { + _consumer.Commit(consumeResult); + _consumer.StoreOffset(consumeResult); + } + } + catch (Exception retryEx) { - _consumer.Commit(consumeResult); - _consumer.StoreOffset(consumeResult); + _logger.LogError(retryEx, "// {Class} // ConsumeMessage // An error occurred while retrying message processing. Offset not committed - message will be reprocessed", GetType().Name); + // Don't commit offset - message will be reprocessed on next consume + // Don't throw - allow consumer to continue processing other messages } _logger.LogError(ex, "// {Class} // ConsumeMessage // An error occurred while consuming messages", GetType().Name); }This approach:
- Prevents offset commit when retry fails (message will be reprocessed)
- Allows the consumer to continue processing subsequent messages
- Logs the retry failure for observability
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs(1 hunks)src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs(1 hunks)src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cs(1 hunks)src/Altinn.Notifications.Integrations/Kafka/Consumers/PastDueOrdersConsumer.cs(1 hunks)
🧰 Additional context used
🧠 Learnings (11)
📓 Common learnings
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1021
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0
Timestamp: 2025-09-16T06:48:40.494Z
Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/PastDueOrdersConsumer.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/PastDueOrdersConsumer.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-15T09:47:14.949Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs:40-54
Timestamp: 2025-10-15T09:47:14.949Z
Learning: In Altinn.Notifications retry consumers (EmailStatusRetryConsumer, SmsStatusRetryConsumer), messages with invalid or unparseable data (null/empty SendOperationResult, deserialization failures) should not be retried. The implementation should log the error and return early without throwing exceptions, as throwing would trigger the base class retry mechanism unintentionally.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/PastDueOrdersConsumer.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-09T10:24:50.624Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1068
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:27-27
Timestamp: 2025-10-09T10:24:50.624Z
Learning: In Altinn.Notifications status consumers (EmailStatusConsumer, SmsStatusConsumer), the base constructor accepts three topic parameters: topicName (to consume from), retryTopicName (for general exceptions - typically the same as topicName for immediate retry), and sendStatusUpdateRetryTopicName (for SendStatusUpdateException - routed to dedicated retry topics like EmailStatusUpdatedRetryTopicName). This implements a two-tier retry strategy: immediate retries via same topic, and threshold-based retries via dedicated retry consumer.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/PastDueOrdersConsumer.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-16T06:48:40.494Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1021
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0
Timestamp: 2025-09-16T06:48:40.494Z
Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/PastDueOrdersConsumer.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-15T08:05:01.433Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs:0-0
Timestamp: 2025-10-15T08:05:01.433Z
Learning: In Altinn.Notifications NotificationStatusConsumerBase.ProcessStatus: ArgumentException and InvalidOperationException are caught and logged with message-only (no exception details). The LogProcessingError method intentionally does not pass the exception parameter to the logger. Tests should verify logging occurred without expecting the exception parameter.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-15T09:58:06.991Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications consumers, using concrete logger types like ILogger<EmailStatusConsumer> instead of the base class generic logger type is the preferred pattern. This provides clearer log categories while still satisfying DI requirements, as the .NET DI container handles the type mapping correctly.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications EmailStatusConsumer, using ILogger<EmailStatusConsumer> instead of the base class generic logger type is intentional and preferred, as it provides clearer log categories while still satisfying DI requirements through the container's type mapping capabilities.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs
📚 Learning: 2025-11-19T08:43:50.344Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1141
File: src/Altinn.Notifications.Persistence/Repository/NotificationRepositoryBase.cs:221-221
Timestamp: 2025-11-19T08:43:50.344Z
Learning: In Altinn.Notifications NotificationRepositoryBase.ReadRecipients: The team has decided not to add a null-guard for the status column when reading from the database. If status is NULL, it indicates data corruption and the code should throw an exception (InvalidCastException from GetFieldValueAsync<string>) to surface the issue immediately, rather than handling it gracefully.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Build, test & analyze
🔇 Additional comments (2)
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cs (1)
95-99: LGTM! Retry validation prevents message loss.The validation of
ProduceAsyncresult ensures that retry publishing failures are detected and prevent offset commits in the base class, aligning with the PR's objective to prevent message loss.src/Altinn.Notifications.Integrations/Kafka/Consumers/PastDueOrdersConsumer.cs (1)
58-62: LGTM! Retry validation consistent with other consumers.The validation correctly checks the
ProduceAsyncresult and throws with an appropriate error message referencing the retry topic. This prevents offset commits when retry publishing fails.
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs
Show resolved
Hide resolved
…tionStatusConsumerBase.cs Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Contributor
Author
|
@coderabbitai review |
Contributor
✅ Actions performedReview triggered.
|
Contributor
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs (1)
95-116: Offset commit gating on retry outcome looks correct; consider resettingmessage/consumeResultdefensively.The new pattern:
- Calls
retryMessageFuncinside a nested try/catch.- Commits & stores the offset only when
retrySucceeded && consumeResult != null.This addresses the core bug where offsets could be committed even when retry failed, which could lead to message loss. That part looks solid.
One subtle improvement: because
messageandconsumeResultare defined outside the loop and not reset per-iteration, exceptions thrown by_consumer.Consumeor by the commit/store calls themselves will cause the retry path to run with the previous message/consumeResult (or an empty string on the very first iteration). That’s not new behavior, but the new retry logic now depends on these values to decide whether to commit.To make this more robust and future‑proof, consider resetting at the top of the loop:
while (!stoppingToken.IsCancellationRequested) { + message = string.Empty; + consumeResult = null; try { consumeResult = _consumer.Consume(stoppingToken); if (consumeResult != null) { message = consumeResult.Message.Value; await processMessageFunc(message); _consumer.Commit(consumeResult); _consumer.StoreOffset(consumeResult); } }This keeps retries and conditional commits strictly tied to the current iteration’s message.
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs (1)
120-124: RetryStatus correctly validatesProduceAsyncresult and surfaces failures with the right topic name.Using the
boolresult fromIKafkaProducer.ProduceAsyncand throwingInvalidOperationExceptionwhen it’sfalseis consistent with the producer contract and ensures republish failures are visible to the base consumer logic. The error message now references_statusUpdatedTopicName, which matches the actual topic used and fixes the earlier mismatch from prior review comments. Based on learnings.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs(1 hunks)src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs(2 hunks)test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs(1 hunks)
🧰 Additional context used
🧠 Learnings (17)
📓 Common learnings
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1148
File: src/Altinn.Notifications.Persistence/Migration/v0.67/01-functions-and-procedures.sql:632-644
Timestamp: 2025-11-12T08:49:13.705Z
Learning: In the Altinn/altinn-notifications repository, files like `src/Altinn.Notifications.Persistence/Migration/v0.67/01-functions-and-procedures.sql` are auto-generated from DbTools and contain many deprecated functions. When reviewing these migration files, only focus on the specific functions relevant to the PR's objective (e.g., for issue #1122 concurrent execution work) rather than reviewing all functions in the file.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cstest/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-09-16T06:48:40.494Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1021
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0
Timestamp: 2025-09-16T06:48:40.494Z
Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cstest/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-10-09T10:24:50.624Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1068
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:27-27
Timestamp: 2025-10-09T10:24:50.624Z
Learning: In Altinn.Notifications status consumers (EmailStatusConsumer, SmsStatusConsumer), the base constructor accepts three topic parameters: topicName (to consume from), retryTopicName (for general exceptions - typically the same as topicName for immediate retry), and sendStatusUpdateRetryTopicName (for SendStatusUpdateException - routed to dedicated retry topics like EmailStatusUpdatedRetryTopicName). This implements a two-tier retry strategy: immediate retries via same topic, and threshold-based retries via dedicated retry consumer.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cstest/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-10-15T09:47:14.949Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs:40-54
Timestamp: 2025-10-15T09:47:14.949Z
Learning: In Altinn.Notifications retry consumers (EmailStatusRetryConsumer, SmsStatusRetryConsumer), messages with invalid or unparseable data (null/empty SendOperationResult, deserialization failures) should not be retried. The implementation should log the error and return early without throwing exceptions, as throwing would trigger the base class retry mechanism unintentionally.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cstest/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-09-15T09:58:06.991Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-15T08:05:01.433Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs:0-0
Timestamp: 2025-10-15T08:05:01.433Z
Learning: In Altinn.Notifications NotificationStatusConsumerBase.ProcessStatus: ArgumentException and InvalidOperationException are caught and logged with message-only (no exception details). The LogProcessingError method intentionally does not pass the exception parameter to the logger. Tests should verify logging occurred without expecting the exception parameter.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cstest/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cstest/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-04-22T14:33:38.561Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 806
File: src/Altinn.Notifications.Persistence/Repository/NotificationDeliveryManifestRepository.cs:0-0
Timestamp: 2025-04-22T14:33:38.561Z
Learning: The team has decided to remove the StatusDescription property from NotificationDeliveryManifest and not implement status parsing in the Altinn.Notifications codebase. Status values should be used as-is without additional parsing or description text.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications consumers, using concrete logger types like ILogger<EmailStatusConsumer> instead of the base class generic logger type is the preferred pattern. This provides clearer log categories while still satisfying DI requirements, as the .NET DI container handles the type mapping correctly.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cssrc/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cstest/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications EmailStatusConsumer, using ILogger<EmailStatusConsumer> instead of the base class generic logger type is intentional and preferred, as it provides clearer log categories while still satisfying DI requirements through the container's type mapping capabilities.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cstest/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-07-16T15:22:50.953Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 940
File: test/Altinn.Notifications.IntegrationTests/Notifications/InstantOrdersController/InstantOrdersControllerTests.cs:506-543
Timestamp: 2025-07-16T15:22:50.953Z
Learning: In the Altinn Notifications system, [Required] attributes on model classes like ShortMessageContentExt, ShortMessageDeliveryDetailsExt, InstantNotificationRecipientExt, and InstantNotificationOrderRequestExt cause deserialization to fail when the provided JSON string lacks the required values. This is different from typical ASP.NET Core behavior where [Required] attributes cause model validation failures after successful deserialization.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs
📚 Learning: 2025-06-13T14:04:57.506Z
Learnt from: SandGrainOne
Repo: Altinn/altinn-notifications PR: 895
File: src/Altinn.Notifications/Program.cs:215-233
Timestamp: 2025-06-13T14:04:57.506Z
Learning: The Altinn.Notifications team intends to replace the current client-secret based Key Vault integration with Workload Identity authentication in a future update; the present code is a temporary measure.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs
📚 Learning: 2025-11-19T08:43:50.344Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1141
File: src/Altinn.Notifications.Persistence/Repository/NotificationRepositoryBase.cs:221-221
Timestamp: 2025-11-19T08:43:50.344Z
Learning: In Altinn.Notifications NotificationRepositoryBase.ReadRecipients: The team has decided not to add a null-guard for the status column when reading from the database. If status is NULL, it indicates data corruption and the code should throw an exception (InvalidCastException from GetFieldValueAsync<string>) to surface the issue immediately, rather than handling it gracefully.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs
📚 Learning: 2025-09-18T10:37:21.188Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1008
File: test/k6/src/data/orders/order-with-reminders-for-organizations.json:34-38
Timestamp: 2025-09-18T10:37:21.188Z
Learning: In Altinn.Notifications test data files, channelSchema enum values can use both "SMS" and "Sms" formats - both are valid in the system, contrary to initial assumptions about strict camel-case requirements.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs
📚 Learning: 2025-10-06T09:01:36.029Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1066
File: src/Altinn.Notifications.Persistence/Migration/v0.58/02-functions-and-procedures.sql:0-0
Timestamp: 2025-10-06T09:01:36.029Z
Learning: In Altinn.Notifications EmailNotificationRepository: When `_alternateid` is provided to `updateemailnotification` function but the corresponding row no longer exists, the system redirects the delivery report to a retry topic and then to the dead delivery reports table (see issue #1017). No fallback to `operationid` should occur in this scenario.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
📚 Learning: 2025-05-23T05:52:42.786Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 861
File: test/Altinn.Notifications.IntegrationTests/Notifications.Persistence/NotificationDeliveryManifestRepositoryTests.cs:165-169
Timestamp: 2025-05-23T05:52:42.786Z
Learning: In NotificationDeliveryManifestRepositoryTests, it's acceptable to bypass status transition logic and directly set order statuses when testing status mapping behavior, as the primary objective is to validate how the mapping logic interprets various order statuses rather than testing the transition logic itself.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs
🔇 Additional comments (1)
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs (1)
73-150: Test assertion on log state will fail; assert on exception message instead.
ProcessEmailDeliveryReport_WhenRetryStatusProducerFails_ThrowsInvalidOperationExceptionverifies that whenProduceAsyncfails, anInvalidOperationExceptionwith message"Failed to republish message to topic..."is thrown and logged.However, your assertion misplaces this expectation. When
KafkaConsumerBase.ConsumeMessagecatches the retry exception, it logs:_logger.LogError(retryEx, "// {Class} // ConsumeMessage // An error occurred while retrying message processing", GetType().Name);In Microsoft.Extensions.Logging, the log
stateparameter is the formatted template text:"// EmailStatusConsumer // ConsumeMessage // An error occurred while retrying message processing". The phrase"Failed to republish message to topic"exists only in the exception'sMessageproperty, not in the state. Your current assertion:It.Is<It.IsAnyType>((v, t) => v.ToString()!.Contains("Failed to republish message to topic")), It.IsAny<InvalidOperationException>(),will not match because the state does not contain that text.
Fix: Assert on the exception message instead:
logger.Verify( e => e.Log( LogLevel.Error, It.IsAny<EventId>(), It.IsAny<It.IsAnyType>(), It.Is<InvalidOperationException>(ex => ex.Message.Contains("Failed to republish message to topic")), It.Is<Func<It.IsAnyType, Exception?, string>>((v, t) => true)), Times.Once);This directly validates that the expected
InvalidOperationExceptionwith the correct message was logged as the exception parameter, independent of the log template wording.⛔ Skipped due to learnings
Learnt from: Ahmed-Ghanam Repo: Altinn/altinn-notifications PR: 1091 File: test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs:0-0 Timestamp: 2025-10-15T08:05:01.433Z Learning: In Altinn.Notifications NotificationStatusConsumerBase.ProcessStatus: ArgumentException and InvalidOperationException are caught and logged with message-only (no exception details). The LogProcessingError method intentionally does not pass the exception parameter to the logger. Tests should verify logging occurred without expecting the exception parameter.Learnt from: Ahmed-Ghanam Repo: Altinn/altinn-notifications PR: 1033 File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91 Timestamp: 2025-09-17T08:45:54.485Z Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.Learnt from: Ahmed-Ghanam Repo: Altinn/altinn-notifications PR: 1091 File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs:40-54 Timestamp: 2025-10-15T09:47:14.949Z Learning: In Altinn.Notifications retry consumers (EmailStatusRetryConsumer, SmsStatusRetryConsumer), messages with invalid or unparseable data (null/empty SendOperationResult, deserialization failures) should not be retried. The implementation should log the error and return early without throwing exceptions, as throwing would trigger the base class retry mechanism unintentionally.Learnt from: Ahmed-Ghanam Repo: Altinn/altinn-notifications PR: 1033 File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91 Timestamp: 2025-09-17T08:45:54.485Z Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.Learnt from: eskebab Repo: Altinn/altinn-notifications PR: 1068 File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:27-27 Timestamp: 2025-10-09T10:24:50.624Z Learning: In Altinn.Notifications status consumers (EmailStatusConsumer, SmsStatusConsumer), the base constructor accepts three topic parameters: topicName (to consume from), retryTopicName (for general exceptions - typically the same as topicName for immediate retry), and sendStatusUpdateRetryTopicName (for SendStatusUpdateException - routed to dedicated retry topics like EmailStatusUpdatedRetryTopicName). This implements a two-tier retry strategy: immediate retries via same topic, and threshold-based retries via dedicated retry consumer.Learnt from: Ahmed-Ghanam Repo: Altinn/altinn-notifications PR: 1021 File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0 Timestamp: 2025-09-16T06:48:40.494Z Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs (1)
100-112: Commit gating after retry looks correct; consider skipping retry when no message was consumedThe new pattern of wrapping
retryMessageFuncin its own try/catch and only committing/storing offsets whenretrySucceeded && consumeResult != nullis a solid fix for avoiding offset commits after failed retries and matches the PR’s intent.One follow-up refinement to consider:
- In cases where
_consumer.Consume(stoppingToken)throws before producing aconsumeResult,consumeResultremainsnullandmessageis juststring.Empty, butretryMessageFunc(message!)is still invoked. That can lead to retry handlers receiving an empty/invalid payload for an exception that wasn’t tied to a concrete message (e.g., broker/network issues).You could guard the retry call itself on having an actual consumed message, which also makes the intent explicit:
catch (Exception ex) { if (consumeResult != null) { var retrySucceeded = false; try { await retryMessageFunc(message); retrySucceeded = true; } catch (Exception retryEx) { _logger.LogError( retryEx, "// {Class} // ConsumeMessage // An error occurred while retrying message processing", GetType().Name); } if (retrySucceeded) { _consumer.Commit(consumeResult); _consumer.StoreOffset(consumeResult); } } _logger.LogError( ex, "// {Class} // ConsumeMessage // An error occurred while consuming messages", GetType().Name); }(With this pattern, the
message!null-forgiving operator can also be dropped, sincemessageis always initialized per iteration.)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs(2 hunks)
🧰 Additional context used
🧠 Learnings (8)
📓 Common learnings
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
📚 Learning: 2025-10-09T10:24:50.624Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1068
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:27-27
Timestamp: 2025-10-09T10:24:50.624Z
Learning: In Altinn.Notifications status consumers (EmailStatusConsumer, SmsStatusConsumer), the base constructor accepts three topic parameters: topicName (to consume from), retryTopicName (for general exceptions - typically the same as topicName for immediate retry), and sendStatusUpdateRetryTopicName (for SendStatusUpdateException - routed to dedicated retry topics like EmailStatusUpdatedRetryTopicName). This implements a two-tier retry strategy: immediate retries via same topic, and threshold-based retries via dedicated retry consumer.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-15T09:47:14.949Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs:40-54
Timestamp: 2025-10-15T09:47:14.949Z
Learning: In Altinn.Notifications retry consumers (EmailStatusRetryConsumer, SmsStatusRetryConsumer), messages with invalid or unparseable data (null/empty SendOperationResult, deserialization failures) should not be retried. The implementation should log the error and return early without throwing exceptions, as throwing would trigger the base class retry mechanism unintentionally.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-15T09:58:06.991Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-16T06:48:40.494Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1021
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0
Timestamp: 2025-09-16T06:48:40.494Z
Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications consumers, using concrete logger types like ILogger<EmailStatusConsumer> instead of the base class generic logger type is the preferred pattern. This provides clearer log categories while still satisfying DI requirements, as the .NET DI container handles the type mapping correctly.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
🔇 Additional comments (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs (1)
75-82: Per-iteration reset of state avoids stale message/offset reuseResetting
messageandconsumeResultat the start of each loop iteration ensures the catch/retry path can never operate on a stale message or offset from a previous iteration, which aligns well with the offset-safety goals of this PR.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs (1)
119-119: Consider logging clarity when retry succeeds.The original error is logged even when
retrySucceededis true (retry recovered from the failure). While this may be intentional for observability, it could be confusing in logs as it reports an error for a successfully recovered message.Consider:
- Adjusting the log message or level when retry succeeds (e.g., "Message processing failed but retry succeeded")
- Or moving this log inside the
if (!retrySucceeded)block to only log when the message is genuinely unrecoverable in this iteration
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs(2 hunks)
🧰 Additional context used
🧠 Learnings (8)
📓 Common learnings
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
📚 Learning: 2025-09-15T09:58:06.991Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-09T10:24:50.624Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1068
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:27-27
Timestamp: 2025-10-09T10:24:50.624Z
Learning: In Altinn.Notifications status consumers (EmailStatusConsumer, SmsStatusConsumer), the base constructor accepts three topic parameters: topicName (to consume from), retryTopicName (for general exceptions - typically the same as topicName for immediate retry), and sendStatusUpdateRetryTopicName (for SendStatusUpdateException - routed to dedicated retry topics like EmailStatusUpdatedRetryTopicName). This implements a two-tier retry strategy: immediate retries via same topic, and threshold-based retries via dedicated retry consumer.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-15T09:47:14.949Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs:40-54
Timestamp: 2025-10-15T09:47:14.949Z
Learning: In Altinn.Notifications retry consumers (EmailStatusRetryConsumer, SmsStatusRetryConsumer), messages with invalid or unparseable data (null/empty SendOperationResult, deserialization failures) should not be retried. The implementation should log the error and return early without throwing exceptions, as throwing would trigger the base class retry mechanism unintentionally.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-16T06:48:40.494Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1021
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0
Timestamp: 2025-09-16T06:48:40.494Z
Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications consumers, using concrete logger types like ILogger<EmailStatusConsumer> instead of the base class generic logger type is the preferred pattern. This provides clearer log categories while still satisfying DI requirements, as the .NET DI container handles the type mapping correctly.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Build, test & analyze
- GitHub Check: build
🔇 Additional comments (3)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs (3)
75-76: LGTM: Clean per-iteration state management.Declaring variables at broader scope and reinitializing them at the start of each iteration prevents state leakage and ensures each message is processed with a clean slate.
Also applies to: 80-81
90-90: Offset commit after successful processing looks correct.Using the
CommitOffsethelper after successful processing ensures consistent offset management across both success and retry-success paths.
101-116: Poison message handling is adequate; infinite reprocessing risk is mitigated.Verification confirms:
Retry consumers handle poison messages gracefully: Both
EmailStatusRetryConsumerandSmsStatusRetryConsumerdetect invalid data (nullSendOperationResult, deserialization failures) via early returns without throwing exceptions. Per the learnings, this prevents unintended retry mechanism triggers.Timeout + dead delivery mechanism prevents infinite loops:
ProcessStatus()checks if retry timeout (StatusUpdatedRetryThresholdSeconds) is exceeded before retrying. Once exceeded,PersistDeadDeliveryReport()saves the message to a dead delivery report and prevents further reprocessing. Additionally,NotificationExpiredExceptiontriggers immediate dead delivery persistence without retry.Exception-based retries are bounded: Messages that throw exceptions are re-published to the retry topic with an incremented attempt count, but they eventually hit the timeout threshold and transition to dead delivery.
The new offset commit logic correctly prevents message loss for transient failures. Poison messages are logged as errors and consumed once (offset committed), while retryable failures eventually drain via the timeout mechanism.
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
Show resolved
Hide resolved
eskebab
force-pushed
the
bug/1097-consumer-offset-commit
branch
from
54216e2 to
f542153
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs (1)
127-131: SimplifyCommitOffsetto rely only onCommit(consumeResult)With
EnableAutoCommit = falseandEnableAutoOffsetStore = false, and given this base does not use_consumer.Commit()without arguments, calling_consumer.StoreOffset(consumeResult)immediately after_consumer.Commit(consumeResult)doesn’t add value and mixes the “immediate commit” and “stored offset” patterns.You can simplify and clarify intent by removing the
StoreOffsetcall:- private void CommitOffset(ConsumeResult<string, string> consumeResult) - { - _consumer.Commit(consumeResult); - _consumer.StoreOffset(consumeResult); - } + private void CommitOffset(ConsumeResult<string, string> consumeResult) + { + _consumer.Commit(consumeResult); + }This matches the manual-commit pattern and avoids suggesting there is any later batch commit depending on stored offsets.
🧹 Nitpick comments (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs (1)
75-123: Retry-gated commit logic looks correct; confirm behavior whenConsumethrowsThe per-iteration reset of
consumeResult/messageand committing only on the happy path or whenretryMessageFuncsucceeds fixes the stale-state issue and avoids committing offsets when both primary processing and the retry path fail, which is what you want for issue #1097.One behavioral nuance to double-check: if
_consumer.Consume(stoppingToken)(or any code beforeconsumeResultis set) throws a non‑OperationCanceledException,consumeResultremainsnull, you log once, and thenreturn, which permanently stops this background service instance. If such errors are intended to be fatal, this is fine; if you expect some of them to be transient, you might instead want a different strategy (e.g., limited retries or letting the exception bubble to a higher-level policy).Minor: inside the retry branch you could avoid the
message!null-forgiveness and any chance of retrying an empty string by passingconsumeResult.Message.Valuedirectly toretryMessageFunc.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs(2 hunks)
🧰 Additional context used
🧠 Learnings (8)
📓 Common learnings
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
📚 Learning: 2025-09-15T09:58:06.991Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-15T09:47:14.949Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs:40-54
Timestamp: 2025-10-15T09:47:14.949Z
Learning: In Altinn.Notifications retry consumers (EmailStatusRetryConsumer, SmsStatusRetryConsumer), messages with invalid or unparseable data (null/empty SendOperationResult, deserialization failures) should not be retried. The implementation should log the error and return early without throwing exceptions, as throwing would trigger the base class retry mechanism unintentionally.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-10-09T10:24:50.624Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1068
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:27-27
Timestamp: 2025-10-09T10:24:50.624Z
Learning: In Altinn.Notifications status consumers (EmailStatusConsumer, SmsStatusConsumer), the base constructor accepts three topic parameters: topicName (to consume from), retryTopicName (for general exceptions - typically the same as topicName for immediate retry), and sendStatusUpdateRetryTopicName (for SendStatusUpdateException - routed to dedicated retry topics like EmailStatusUpdatedRetryTopicName). This implements a two-tier retry strategy: immediate retries via same topic, and threshold-based retries via dedicated retry consumer.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-16T06:48:40.494Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1021
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0
Timestamp: 2025-09-16T06:48:40.494Z
Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications consumers, using concrete logger types like ILogger<EmailStatusConsumer> instead of the base class generic logger type is the preferred pattern. This provides clearer log categories while still satisfying DI requirements, as the .NET DI container handles the type mapping correctly.
Applied to files:
src/Altinn.Notifications.Integrations/Kafka/Consumers/KafkaConsumerBase.cs
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Build, test & analyze
- GitHub Check: Analyze (csharp, autobuild)
eskebab
force-pushed
the
bug/1097-consumer-offset-commit
branch
from
f542153 to
d1693bd
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs (2)
228-304: Consider verifying consumer resilience after producer failure.Based on the PR objectives to prevent "consumer crashes" when committing offsets on retry failure, consider adding an assertion to verify the consumer continues to operate after the producer failure. For example, you could publish a second valid message after the failure and verify it's processed successfully, demonstrating the consumer hasn't crashed.
229-229: Consider clarifying the test method name.The test name
ConsumeRetryMessage_WhenRetryStatusProducerFails_ThrowsInvalidOperationExceptionmight imply the test method itself expects an exception to be thrown. However, the test actually verifies that an InvalidOperationException is thrown internally byRetryStatusand handled gracefully (logged without crashing the consumer). Consider a name likeConsumeRetryMessage_WhenRetryStatusProducerFails_LogsErrorAndContinuesto better reflect the expected behavior.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs(1 hunks)
🧰 Additional context used
🧠 Learnings (10)
📓 Common learnings
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
📚 Learning: 2025-10-15T08:05:01.433Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs:0-0
Timestamp: 2025-10-15T08:05:01.433Z
Learning: In Altinn.Notifications NotificationStatusConsumerBase.ProcessStatus: ArgumentException and InvalidOperationException are caught and logged with message-only (no exception details). The LogProcessingError method intentionally does not pass the exception parameter to the logger. Tests should verify logging occurred without expecting the exception parameter.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-10-15T09:47:14.949Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs:40-54
Timestamp: 2025-10-15T09:47:14.949Z
Learning: In Altinn.Notifications retry consumers (EmailStatusRetryConsumer, SmsStatusRetryConsumer), messages with invalid or unparseable data (null/empty SendOperationResult, deserialization failures) should not be retried. The implementation should log the error and return early without throwing exceptions, as throwing would trigger the base class retry mechanism unintentionally.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-10-09T10:24:50.624Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1068
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:27-27
Timestamp: 2025-10-09T10:24:50.624Z
Learning: In Altinn.Notifications status consumers (EmailStatusConsumer, SmsStatusConsumer), the base constructor accepts three topic parameters: topicName (to consume from), retryTopicName (for general exceptions - typically the same as topicName for immediate retry), and sendStatusUpdateRetryTopicName (for SendStatusUpdateException - routed to dedicated retry topics like EmailStatusUpdatedRetryTopicName). This implements a two-tier retry strategy: immediate retries via same topic, and threshold-based retries via dedicated retry consumer.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-16T06:48:40.494Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1021
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0
Timestamp: 2025-09-16T06:48:40.494Z
Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-15T09:58:06.991Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications consumers, using concrete logger types like ILogger<EmailStatusConsumer> instead of the base class generic logger type is the preferred pattern. This provides clearer log categories while still satisfying DI requirements, as the .NET DI container handles the type mapping correctly.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications EmailStatusConsumer, using ILogger<EmailStatusConsumer> instead of the base class generic logger type is intentional and preferred, as it provides clearer log categories while still satisfying DI requirements through the container's type mapping capabilities.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
🧬 Code graph analysis (1)
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs (5)
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cs (8)
Task(51-54)Task(61-79)Task(86-86)Task(93-100)Task(135-150)Task(157-182)Task(189-204)UpdateStatusRetryMessage(109-128)src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs (1)
EmailStatusRetryConsumer(18-67)src/Altinn.Notifications.Core/Models/UpdateStatusRetryMessage.cs (1)
Serialize(33-33)test/Altinn.Notifications.IntegrationTests/Utils/KafkaUtil.cs (1)
KafkaUtil(7-94)test/Altinn.Notifications.IntegrationTests/Utils/IntegrationTestUtil.cs (1)
IntegrationTestUtil(7-123)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Build, test & analyze
- GitHub Check: Analyze (csharp, autobuild)
...ests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
Show resolved
Hide resolved
eskebab
force-pushed
the
bug/1097-consumer-offset-commit
branch
from
2a1571b to
10425f3
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs (1)
228-307: Test implementation looks good and comprehensive.The test correctly verifies the producer failure scenario and includes all necessary assertions. The past review comment about adding
deadDeliveryReportServiceverification has been addressed (line 295).Minor observations:
Timeout inconsistency: This test uses a 10-second timeout (line 304) while other tests in the file use 15 seconds (lines 63, 142, 212). Consider using 15 seconds for consistency unless there's a specific reason for the shorter timeout.
Test name clarity: The test name
ThrowsInvalidOperationExceptionsuggests the exception propagates to the caller, but the exception is actually caught and logged internally by the consumer's error handling. The test verifies proper error handling rather than exception propagation. This is correct behavior based on the PR objectives.Consider this diff for timeout consistency:
}, - TimeSpan.FromSeconds(10)); + TimeSpan.FromSeconds(15)); await emailStatusConsumer.StopAsync(CancellationToken.None);
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs(1 hunks)
🧰 Additional context used
🧠 Learnings (11)
📓 Common learnings
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
📚 Learning: 2025-10-15T09:47:14.949Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs:40-54
Timestamp: 2025-10-15T09:47:14.949Z
Learning: In Altinn.Notifications retry consumers (EmailStatusRetryConsumer, SmsStatusRetryConsumer), messages with invalid or unparseable data (null/empty SendOperationResult, deserialization failures) should not be retried. The implementation should log the error and return early without throwing exceptions, as throwing would trigger the base class retry mechanism unintentionally.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-10-15T08:05:01.433Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1091
File: test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusConsumerBaseTests.cs:0-0
Timestamp: 2025-10-15T08:05:01.433Z
Learning: In Altinn.Notifications NotificationStatusConsumerBase.ProcessStatus: ArgumentException and InvalidOperationException are caught and logged with message-only (no exception details). The LogProcessingError method intentionally does not pass the exception parameter to the logger. Tests should verify logging occurred without expecting the exception parameter.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-10-09T10:24:50.624Z
Learnt from: eskebab
Repo: Altinn/altinn-notifications PR: 1068
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:27-27
Timestamp: 2025-10-09T10:24:50.624Z
Learning: In Altinn.Notifications status consumers (EmailStatusConsumer, SmsStatusConsumer), the base constructor accepts three topic parameters: topicName (to consume from), retryTopicName (for general exceptions - typically the same as topicName for immediate retry), and sendStatusUpdateRetryTopicName (for SendStatusUpdateException - routed to dedicated retry topics like EmailStatusUpdatedRetryTopicName). This implements a two-tier retry strategy: immediate retries via same topic, and threshold-based retries via dedicated retry consumer.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: In Altinn.Notifications, the IKafkaProducer.ProduceAsync method includes internal try-catch handling and returns false if any exception occurs during publishing. This means external callers can rely on the boolean return value rather than needing explicit exception handling. Additionally, once GetNewNotifications() claims a batch of SMS notifications (atomically changing status from "New" to "Sending" via SQL), the design expects those claimed notifications to be processed to completion or reset to "New" status, rather than supporting mid-batch cancellation.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-16T06:48:40.494Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1021
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:0-0
Timestamp: 2025-09-16T06:48:40.494Z
Learning: In Altinn.Notifications SMS processing (SmsNotificationService.SendNotifications), when GetNewNotifications() retrieves SMS notifications, their status is atomically changed from "New" to "Sending". If publishing to Kafka fails (ProduceAsync returns false or throws), the status must be reset to "New" via UpdateSendStatus to allow retry processing in future batches. This prevents notifications from getting stuck in "Sending" state.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-15T09:58:06.991Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusConsumerBase.cs:158-161
Timestamp: 2025-09-15T09:58:06.991Z
Learning: In Altinn.Notifications status consumers, Kafka messages for status updates contain only references and send-status information (non-sensitive data), so logging the full Kafka message payload is acceptable according to maintainer Ahmed-Ghanam.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications consumers, using concrete logger types like ILogger<EmailStatusConsumer> instead of the base class generic logger type is the preferred pattern. This provides clearer log categories while still satisfying DI requirements, as the .NET DI container handles the type mapping correctly.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-02T09:27:21.022Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1015
File: src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusConsumer.cs:28-35
Timestamp: 2025-09-02T09:27:21.022Z
Learning: In Altinn.Notifications EmailStatusConsumer, using ILogger<EmailStatusConsumer> instead of the base class generic logger type is intentional and preferred, as it provides clearer log categories while still satisfying DI requirements through the container's type mapping capabilities.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-09-17T08:45:54.485Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1033
File: src/Altinn.Notifications.Core/Services/SmsNotificationService.cs:71-91
Timestamp: 2025-09-17T08:45:54.485Z
Learning: The KafkaProducer.ProduceAsync method in Altinn.Notifications includes comprehensive internal error handling with try-catch blocks that catch both ProduceException and general Exception types. It returns false on any error condition (exceptions or delivery failures) and only returns true when the message is successfully persisted. This means callers can rely on the boolean return value rather than needing additional exception handling.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
📚 Learning: 2025-10-06T09:01:36.029Z
Learnt from: Ahmed-Ghanam
Repo: Altinn/altinn-notifications PR: 1066
File: src/Altinn.Notifications.Persistence/Migration/v0.58/02-functions-and-procedures.sql:0-0
Timestamp: 2025-10-06T09:01:36.029Z
Learning: In Altinn.Notifications EmailNotificationRepository: When `_alternateid` is provided to `updateemailnotification` function but the corresponding row no longer exists, the system redirects the delivery report to a retry topic and then to the dead delivery reports table (see issue #1017). No fallback to `operationid` should occur in this scenario.
Applied to files:
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs
🧬 Code graph analysis (1)
test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/NotificationStatusRetryConsumerBaseTests.cs (4)
src/Altinn.Notifications.Integrations/Kafka/Consumers/NotificationStatusRetryConsumerBase.cs (8)
Task(51-54)Task(61-79)Task(86-86)Task(93-100)Task(135-150)Task(157-182)Task(189-204)UpdateStatusRetryMessage(109-128)src/Altinn.Notifications.Integrations/Kafka/Consumers/EmailStatusRetryConsumer.cs (1)
EmailStatusRetryConsumer(18-67)test/Altinn.Notifications.IntegrationTests/Utils/KafkaUtil.cs (1)
KafkaUtil(7-94)test/Altinn.Notifications.IntegrationTests/Utils/IntegrationTestUtil.cs (1)
IntegrationTestUtil(7-123)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Analyze (csharp, autobuild)
- GitHub Check: Build, test & analyze
|
Ahmed-Ghanam
requested changes
Nov 25, 2025
Comment on lines
+80
to
+81
| message = string.Empty; | ||
| consumeResult = null; |
Contributor
There was a problem hiding this comment.
We can reduce duplicates by introducing as first line under the while loop:
string message = string.Empty;
ConsumeResult<string, string>? consumeResult = null;| } | ||
| } | ||
|
|
||
| private void CommitOffset(ConsumeResult<string, string> consumeResult) |
Contributor
There was a problem hiding this comment.
Write kindly simple documentation to describe HandleProcessingFailure, TryRetryMessage and CommitOffset
| CommitOffset(consumeResult); | ||
| } | ||
|
|
||
| _logger.LogError(ex, "// {Class} // ConsumeMessage // An error occurred while consuming messages", GetType().Name); |
Contributor
There was a problem hiding this comment.
Why is it necessary to log an error even when the message has been successfully processed by the retry mechanism? Is this because the primary handler failed on the initial attempt? If so, why should we still log an error when the retry mechanism ultimately handled the message successfully?
| } | ||
|
|
||
| [Fact] | ||
| public async Task ProcessEmailDeliveryReport_WhenRetryStatusProducerFails_ThrowsInvalidOperationException() |
Contributor
There was a problem hiding this comment.
Test name says ThrowsInvalidOperationException but the test does not assert a thrown exception;
it asserts logging. A better name could be: ConsumeRetryMessage_WhenRetryStatusProducerFails_ThrowsInvalidOperationException
| } | ||
|
|
||
| [Fact] | ||
| public async Task ConsumeRetryMessage_WhenRetryStatusProducerFails_ThrowsInvalidOperationException() |
Contributor
There was a problem hiding this comment.
Test name says ThrowsInvalidOperationException but the test does not assert a thrown exception;
it asserts logging. A better name could be: ConsumeRetryMessage_WhenRetryStatusProducerFails_ThrowsInvalidOperationException or • ConsumeRetryMessage_WhenRetryPublishFails_IncrementsAttemptsAndLogsInvalidOperationException
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Kafka consumer offset commit on retry failure risks message loss
Description
The current implementation of KafkaConsumerBase.ConsumeMessage has systemic issues affecting all current and future consumers, leading to potential message loss and consumer crashes.
Related Issue(s)
Verification
Documentation
Summary by CodeRabbit
Bug Fixes
Tests
✏️ Tip: You can customize this high-level summary in your review settings.