krishvishal commented on code in PR #4502:
URL: https://github.com/apache/iggy/pull/4502#discussion_r4259829870
##########
gateways/kafka/src/bridge/iggy_bridge/produce.rs:
##########
@@ -19,58 +19,171 @@
//!
//! Separate from `fetch.rs` so the two never edit one file.
+use std::collections::HashMap;
use std::future::Future;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
use iggy::prelude::{IggyError, IggyMessage, MessageClient, Partitioning};
use tokio::sync::Semaphore;
-use tokio::time::{Instant, timeout};
+use tokio::time::{Instant, timeout, timeout_at};
+use super::fetch::LazyClient;
use super::{IggyBridge, SLOT_LIMIT, TopicTarget, in_slot, permit_by};
use crate::bridge::error::BridgeError;
/// Frame bytes besides the messages: request header, batch header and ids.
const SEND_FRAME_ROOM: u64 = 4 * 1024;
+/// Produce sessions open at once, each an Iggy connection.
+const PRODUCE_SESSIONS: usize = 256;
+
+/// A free session unused this long closes when a new producer opens one.
+const SESSION_IDLE: Duration = Duration::from_secs(600);
+
+/// Who sends a Produce request. Each producer gets an Iggy session of its own.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum Producer {
+ /// The producer id of the request's record batches.
+ Id(i64),
+ /// The Kafka connection, for a producer without an id.
+ Connection(u64),
+}
+
+/// One producer's Iggy client, and one send at a time on it, so its sends
reach Iggy in order.
+pub struct ProduceSession {
+ client: LazyClient,
+ turn: Arc<Semaphore>,
+}
+
+/// The Produce sessions, one per producer, so different producers send at
once.
+pub(super) struct ProducePool {
+ sessions: Mutex<HashMap<Producer, Held>>,
+ capacity: usize,
+}
+
+/// A session, and when a request last took it.
+struct Held {
+ session: Arc<ProduceSession>,
+ used: Instant,
+}
+
+impl ProducePool {
+ pub(super) fn new() -> Self {
+ Self::with_capacity(PRODUCE_SESSIONS)
+ }
+
+ fn with_capacity(capacity: usize) -> Self {
+ Self {
+ sessions: Mutex::new(HashMap::new()),
+ capacity,
+ }
+ }
+
+ /// The session of `producer`, opened if it has none. `None` when the pool
is full and every
+ /// session is busy.
+ ///
+ /// Opening one first drops the free sessions idle past [`SESSION_IDLE`],
then the least
+ /// recently used free one if the pool is still full.
+ fn take(&self, producer: Producer) -> Option<Arc<ProduceSession>> {
+ let now = Instant::now();
+ let mut sessions =
self.sessions.lock().unwrap_or_else(PoisonError::into_inner);
+ if let Some(held) = sessions.get_mut(&producer) {
+ held.used = now;
+ return Some(Arc::clone(&held.session));
+ }
+ sessions
+ .retain(|_, held| !(held.is_free() &&
now.duration_since(held.used) >= SESSION_IDLE));
+ if sessions.len() >= self.capacity {
+ let oldest = sessions
+ .iter()
+ .filter(|(_, held)| held.is_free())
+ .min_by_key(|(_, held)| held.used)
+ .map(|(producer, _)| *producer)?;
+ sessions.remove(&oldest);
Review Comment:
Done: logout on close, a full pool shares one client instead of closing busy
sessions, sign-ins are capped, and both limits are env vars now.
##########
gateways/kafka/src/bridge/iggy_bridge/produce.rs:
##########
@@ -191,4 +304,91 @@ mod tests {
Err(BridgeError::SendLost(IggyError::Disconnected))
));
}
+
+ #[tokio::test]
+ async fn given_one_producer_when_taken_twice_should_get_one_session() {
Review Comment:
Added those tests, and main now closes the bridge on shutdown.
--
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]
