Fix to Producer Busy error on reconnect

src/connection.rs

@@ -431,9 +431,11 @@ impl<Exe: Executor> ConnectionSender<Exe> {
         producer_id: u64,
         producer_name: Option<String>,
         options: ProducerOptions,
+        epoch: u64,
+        user_provided_producer_name: bool,
     ) -> Result<proto::CommandProducerSuccess, ConnectionError> {
         let request_id = self.request_id.get();
-        let msg = messages::create_producer(topic, producer_name, producer_id, request_id, options);
+        let msg = messages::create_producer(topic, producer_name, producer_id, request_id, options, epoch, user_provided_producer_name);
         self.send_message(msg, RequestKey::RequestId(request_id), |resp| {
             resp.command.producer_success
         })
@@ -1227,6 +1229,8 @@ pub(crate) mod messages {
         producer_id: u64,
         request_id: u64,
         options: ProducerOptions,
+        epoch: u64,
+        user_provided_producer_name: bool,
     ) -> Message {
         Message {
             command: proto::BaseCommand {
@@ -1235,6 +1239,7 @@ pub(crate) mod messages {
                     topic,
                     producer_id,
                     request_id,
+                    user_provided_producer_name: Some(user_provided_producer_name),
                     producer_name,
                     encrypted: options.encrypted,
                     metadata: options
@@ -1247,6 +1252,7 @@ pub(crate) mod messages {
                         .collect(),
                     schema: options.schema,
                     producer_access_mode: options.access_mode,
+                    epoch: Some(epoch),
                     ..Default::default()
                 }),
                 ..Default::default()

src/producer.rs

@@ -433,6 +433,8 @@ struct TopicProducer<Exe: Executor> {
     batch: Option<Mutex<Batch>>,
     compression: Option<Compression>,
     options: ProducerOptions,
+    epoch: std::sync::atomic::AtomicU64,
+    user_provided_producer_name: bool,
 }
 
 impl<Exe: Executor> TopicProducer<Exe> {
@@ -455,6 +457,8 @@ impl<Exe: Executor> TopicProducer<Exe> {
         let batch_byte_size = options.batch_byte_size;
         let compression = options.compression.clone();
         let mut connection = client.manager.get_connection(&addr).await?;
+        let epoch = std::sync::atomic::AtomicU64::new(0);
+        let user_provided_producer_name = name.is_some();
 
         let producer_name = retry_create_producer(
             &client,
@@ -464,6 +468,8 @@ impl<Exe: Executor> TopicProducer<Exe> {
             producer_id,
             name,
             &options,
+            &epoch,
+            user_provided_producer_name,
         )
         .await?;
 
@@ -481,6 +487,8 @@ impl<Exe: Executor> TopicProducer<Exe> {
             batch,
             compression,
             options,
+            epoch,
+            user_provided_producer_name,
         })
     }
 
@@ -727,6 +735,8 @@ impl<Exe: Executor> TopicProducer<Exe> {
             self.id,
             Some(self.name.clone()),
             &self.options,
+            &self.epoch,
+            self.user_provided_producer_name,
         )
         .await?;
 

src/retry_op.rs

@@ -156,6 +156,8 @@ pub async fn retry_create_producer<Exe: Executor>(
     producer_id: u64,
     producer_name: Option<String>,
     options: &ProducerOptions,
+    epoch: &std::sync::atomic::AtomicU64,
+    user_provided_producer_name: bool,
 ) -> Result<String, Error> {
     *connection = client.manager.get_connection(&addr).await?;
     let mut current_retries = 0u32;
@@ -174,6 +176,8 @@ pub async fn retry_create_producer<Exe: Executor>(
                 producer_id,
                 producer_name.clone(),
                 options.clone(),
+                epoch.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
+                user_provided_producer_name
             )
             .await
         {