Kafka async and logging - #2378
Open
pindo696 wants to merge 4 commits into
Open
Conversation
Reviewer's GuideRefactors Kafka messaging to be fully async end-to-end, tightening logging and error handling, and adjusts recalc batching and notification flows to await Kafka sends explicitly while preserving existing control-flow semantics. Sequence diagram for async evaluation and Kafka messagingsequenceDiagram
participant InventoryProcessor
participant GrouperQueue
participant EvaluatorProcessor
participant KafkaPayloadTracker
participant KafkaRemediations
participant KafkaNotifications
participant KafkaInventoryViews
InventoryProcessor->>InventoryProcessor: _process_upload(msg)
InventoryProcessor->>InventoryProcessor: _send_for_evaluation(org_id, inventory_id, request_id, reporter, timestamp, import_status)
InventoryProcessor->>GrouperQueue: grouper.send(msg, key=org_id)
G rouperQueue-->>InventoryProcessor: await complete
GrouperQueue->>GrouperQueue: _send_for_evaluation(item, org_id, inventory_id, msg)
GrouperQueue->>KafkaPayloadTracker: send_msg_to_payload_tracker(payload_tracker, msg, status="processing")
KafkaPayloadTracker-->>GrouperQueue: await complete
GrouperQueue->>EvaluatorProcessor: evaluator.send(msg)
EvaluatorProcessor-->>GrouperQueue: await complete
EvaluatorProcessor->>EvaluatorProcessor: _evaluate_system(...)
EvaluatorProcessor->>KafkaRemediations: send_remediations_update(remediations_results, inventory_id, fixable_sys_vuln_rows)
KafkaRemediations-->>EvaluatorProcessor: await complete
EvaluatorProcessor->>KafkaNotifications: send_notifications(evaluator_results, new_system_vulns, [], [], rh_account_id, org_id)
KafkaNotifications-->>EvaluatorProcessor: await complete
EvaluatorProcessor->>KafkaInventoryViews: send_inventory_views(inventory_views_results, request_id, inventory_id, ...)
KafkaInventoryViews-->>EvaluatorProcessor: await complete
EvaluatorProcessor->>KafkaPayloadTracker: send_msg_to_payload_tracker(payload_tracker, msg, status="success" | "error")
KafkaPayloadTracker-->>EvaluatorProcessor: await complete
Sequence diagram for async recalc batch send with semaphoresequenceDiagram
participant RecalcCaller as RecalcAccounts_or_VmaasSync
participant BATCH_SEMAPHORE
participant Helper as _send_recalc_batch_and_release
participant EvaluatorQueue as EVALUATOR_QUEUE
loop while rows
RecalcCaller->>BATCH_SEMAPHORE: acquire()
BATCH_SEMAPHORE-->>RecalcCaller: acquired
RecalcCaller->>RecalcCaller: _create_kafka_msg(rows)
RecalcCaller->>Helper: _send_recalc_batch_and_release(msgs)
activate Helper
Helper->>EvaluatorQueue: send_list(msgs)
EvaluatorQueue-->>Helper: await complete
Helper->>BATCH_SEMAPHORE: release()
deactivate Helper
end
RecalcCaller->>RecalcCaller: asyncio.gather(*send_tasks) (vmaas_sync) / run_until_complete(...) (admin_handler)
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Contributor
SC Environment Impact AssessmentOverall Impact: 🟡 MEDIUM View full reportSummary
Detailed Findings🟡 MEDIUM ImpactKafka topic configuration change detected
Kafka topic configuration change detected
Required Actions
This assessment was automatically generated. Please review carefully and consult with the ROSA Core team for critical/high impact changes. |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- In
RecalcAccounts.handle_put,total_scheduledis now overwritten withlen(msgs)on each loop iteration instead of accumulated as before; this changes the semantics from a total count to just the last batch size and is likely a regression compared to the previous behavior. - In
utils.send_msg_to_payload_tracker, theKafkaErroris fully swallowed after logging; if payload-tracker failures should be observable upstream, consider surfacing a metric or returning an error flag so callers can react instead of always treating the flow as successful.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `RecalcAccounts.handle_put`, `total_scheduled` is now overwritten with `len(msgs)` on each loop iteration instead of accumulated as before; this changes the semantics from a total count to just the last batch size and is likely a regression compared to the previous behavior.
- In `utils.send_msg_to_payload_tracker`, the `KafkaError` is fully swallowed after logging; if payload-tracker failures should be observable upstream, consider surfacing a metric or returning an error flag so callers can react instead of always treating the flow as successful.
## Individual Comments
### Comment 1
<location path="vmaas_sync/vmaas_sync.py" line_range="389-393" />
<code_context>
- future = EVALUATOR_QUEUE.send_list(msgs, loop=loop)
- future.add_done_callback(lambda x: BATCH_SEMAPHORE.release())
- futures.append(future)
+ send_tasks.append(_send_recalc_batch_and_release(msgs))
- if futures:
- LOGGER.info("Waiting for %s Kafka send operations to complete", len(futures))
- loop.run_until_complete(asyncio.gather(*futures))
+ if send_tasks:
+ LOGGER.info("Waiting for %s Kafka send operations to complete", len(send_tasks))
+ loop.run_until_complete(asyncio.gather(*send_tasks))
LOGGER.info("%s systems scheduled for re-evaluation", total_scheduled)
</code_context>
<issue_to_address>
**issue (bug_risk):** Potential deadlock: semaphore is acquired in the loop but only released inside tasks that are scheduled after the loop terminates.
Because each iteration only appends the coroutine to `send_tasks` and doesn’t schedule it, no `_send_recalc_batch_and_release` call runs until after the loop finishes and `asyncio.gather(*send_tasks)` is invoked. Once the semaphore permits are exhausted, `loop.run_until_complete(BATCH_SEMAPHORE.acquire())` will block indefinitely, since no running task can release the semaphore. Consider either scheduling the send immediately (e.g. `send_tasks.append(loop.create_task(_send_recalc_batch_and_release(msgs)))`) or restoring the per-iteration send pattern so releases can occur while the loop is still acquiring.
</issue_to_address>
### Comment 2
<location path="notificator/notificator_queue.py" line_range="110-113" />
<code_context>
}
]
- def _send_kafka_notif(
+ async def _send_kafka_notif(
self,
</code_context>
<issue_to_address>
**suggestion:** The `loop` parameter to `_send_kafka_notif` is now unused and can be removed.
Since `_send_kafka_notif` is now async and uses `await self.notifications_topic.send(msg)`, the `loop` parameter is unused. Please remove it from the signature and update callers to avoid confusion about which event loop is in play.
Suggested implementation:
```python
async def _send_kafka_notif(
self,
org_id: str,
inventory_id: str,
msg["org_id"] = org_id
```
```python
```
1. Remove the `loop` argument at all call sites of `_send_kafka_notif` in this file and elsewhere in the codebase. For example, change calls like:
- `await self._send_kafka_notif(loop, org_id, inventory_id, msg)` to `await self._send_kafka_notif(org_id, inventory_id, msg)`.
2. If the `_send_kafka_notif` signature uses a typed or defaulted `loop` parameter (e.g. `loop: AbstractEventLoop` or `loop: Optional[AbstractEventLoop] = None`), delete that entire parameter definition from the function signature to keep it consistent with the updated callers.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Contributor
Author
|
/retest |
pindo696
force-pushed
the
kafka-async-and-logging
branch
from
June 17, 2026 12:08
517046f to
36d09c9
Compare
Member
|
Needs rebase pls |
RHINENG-24284
RHINENG-24284
RE_EVALUATION_KAFKA_BATCHES limits number of tasks in flight, we need to schedule them for processing midrun inside while loop allowing multiple tasks in flight. RHINENG-24284
pindo696
force-pushed
the
kafka-async-and-logging
branch
from
July 27, 2026 14:32
36d09c9 to
636ee84
Compare
Changes that were lost during merge conflicts resolution.
pindo696
force-pushed
the
kafka-async-and-logging
branch
from
July 28, 2026 09:03
a276f5b to
f75058b
Compare
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
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.
Refactoring sending kafka async messages and improved logging,
Updated tests,
RHINENG-24284.
(Hope something has not been lost during the push)
Secure Coding Practices Checklist GitHub Link
Secure Coding Checklist
Summary by Sourcery
Convert Kafka message sending across the service to fully async usage with improved error logging and semaphore handling for batch operations.
Bug Fixes:
Enhancements:
Tests: