ethanlin01x commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3855694794
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -91,6 +91,47 @@ pub enum AutoCommitAfter {
ConsumingEveryNthMessage(u32),
}
+/// A cheap, cloneable view of the state shared with an [`IggyConsumer`].
+///
+/// Consuming borrows the consumer as `&mut` for the whole run, so reading its
getters
+/// concurrently means sharing it behind a lock and then waiting on that lock.
This view
+/// carries the same shared state and needs neither.
+#[derive(Clone)]
+pub struct IggyConsumerState {
+ current_partition_id: Arc<AtomicU32>,
+ last_consumed_offsets: Arc<DashMap<u32, AtomicU64>>,
+ last_stored_offsets: Arc<DashMap<u32, AtomicU64>>,
+}
+
+impl IggyConsumerState {
+ fn new() -> Self {
+ Self {
+ current_partition_id: Arc::new(AtomicU32::new(0)),
+ last_consumed_offsets: Arc::new(DashMap::new()),
+ last_stored_offsets: Arc::new(DashMap::new()),
+ }
+ }
+
+ /// Returns the current partition ID of the consumer.
+ pub fn partition_id(&self) -> u32 {
Review Comment:
Doc updated in 984494952 to say these are independent loads, not a snapshot.
##########
foreign/python/src/consumer.rs:
##########
@@ -44,10 +44,16 @@ use crate::receive_message::ReceiveMessage;
/// A Python class representing the Iggy consumer.
/// It provides asynchronous functionality through the contained runtime.
+// `inner` stays locked for the whole duration of a consumption run, so the
synchronous
Review Comment:
Both moved onto IggyConsumerState in b0d3c20c0, no lock taken from callbacks
anymore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
