Clean up #3693
Clean up #3693
Conversation
📝 WalkthroughWalkthroughThe pull request introduces significant modifications to the Changes
Possibly related PRs
Suggested reviewers
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Outside diff range and nitpick comments (3)
src/workflows/airqo_etl_utils/message_broker_utils.py (3)
Line range hint
360-426: Logical error in loop condition withinconsume_from_topicThe loop condition in the
consume_from_topicmethod may not behave as expected:while streaming or (message_count < max_messages if max_messages else True):When
max_messagesisNoneandstreamingisFalse, the condition simplifies to:while False or True # which is TrueThis causes the loop to run indefinitely, even when
streamingisFalseandmax_messagesis not set. This might not be the intended behavior.Consider revising the loop condition to properly handle scenarios where
max_messagesisNone:- while streaming or (message_count < max_messages if max_messages else True): + while streaming or (max_messages is not None and message_count < max_messages):This adjustment ensures that:
- If
streamingisTrue, the loop runs indefinitely.- If
max_messagesis specified, the loop runs untilmessage_countreachesmax_messages.- If neither
streamingisTruenormax_messagesis specified, the loop exits appropriately.
Line range hint
399-407: Incorrect offset assignment when seeking to a specific offsetIn the
consume_from_topicmethod, when anoffsetis provided, the code attempts to assign partitions with the current offsets instead of the specifiedoffset:partitions = [ TopicPartition(topic, p.partition, p.offset) for p in consumer.assignment() ]Using
p.offsetdoes not set the partitions to the desiredoffset. This might prevent the consumer from starting at the intended position.To resolve this, set the
offsetparameter to the providedoffsetvalue:if offset is not None: logger.info(f"Seeking to offset {offset} for all partitions...") partitions = [ - TopicPartition(topic, p.partition, p.offset) + TopicPartition(topic, p.partition, offset) for p in consumer.assignment() ] consumer.assign(partitions)This change ensures that the consumer begins consuming from the specified
offsetfor all partitions.
Line range hint
388-395: Redundant calls toconsumer.seek()after assignmentWhen
from_beginningisTrue, the code assigns partitions withoffset=0and then callsconsumer.seek(partition)for each partition:consumer.assign(partitions) for partition in partitions: consumer.seek(partition)However, since the partitions have already been assigned with
offset=0, callingconsumer.seek(partition)without specifying an offset may overwrite the intended offset or have no effect, depending on the consumer configuration.It's more reliable to set the desired offset during assignment and omit the
seekcalls:if from_beginning: logger.info("Seeking to the beginning of all partitions...") partitions = [ TopicPartition(topic, p.partition, offset=0) for p in consumer.assignment() ] - consumer.assign(partitions) - for partition in partitions: - consumer.seek(partition) + consumer.assign(partitions)This ensures that the consumer starts from the beginning of each partition without unnecessary redundancy.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (1)
- src/workflows/airqo_etl_utils/message_broker_utils.py (1 hunks)
🧰 Additional context used
📓 Learnings (1)
src/workflows/airqo_etl_utils/message_broker_utils.py (1)
Learnt from: NicholasTurner23 PR: airqo-platform/AirQo-api#3683 File: src/workflows/airqo_etl_utils/message_broker_utils.py:271-302 Timestamp: 2024-10-17T21:49:16.098Z Learning: In the `MessageBrokerUtils` class, messages are flushed in the caller method after invoking `_send_message`, so it's not necessary to call `producer.flush()` within the `_send_message` method.
🔇 Additional comments (1)
src/workflows/airqo_etl_utils/message_broker_utils.py (1)
360-360: Good practice: Flushing the producer to ensure all messages are sentThe addition of
producer.flush()at the end of thepublish_to_topicmethod ensures that all buffered messages are sent before the producer is closed. This guarantees that no messages are lost.
Description
Just some clean up.
Summary by CodeRabbit
New Features
Improvements