feat: support pre-warming PartitionClient

Adds a refresh_leader() method to force the PartitionClient through the
discovery process.

This will ensure a cached connection to the leader is (likely) already
established by the time the first produce() call is issued, preventing
the first produce() calls (multiple of which may be queued up on the
discovery mutex) from experiencing excessively high latency.

Author: domodwyer

src/client/partition.rs

@@ -115,6 +115,17 @@ impl PartitionClient {
         self.partition
     }
 
+    /// Re-run the leader discovery process for this client, and establish a
+    /// connection to the leader.
+    pub async fn refresh_leader(&self) -> Result<()> {
+        // Remove the current broker connection, if any.
+        *self.current_broker.lock().await = None;
+        // And acquire a new one, forcing discovery and establishing a cached
+        // connection to the leader.
+        self.get().await?;
+        Ok(())
+    }
+
     /// Produce a batch of records to the partition
     pub async fn produce(
         &self,