avoid using boxed partitioner

Author: mlowicki

src/client.rs

@@ -221,9 +221,18 @@ impl<C: ClientContext> Client<C> {
         native_config: NativeClientConfig,
         rd_kafka_type: RDKafkaType,
         context: C,
+    ) -> KafkaResult<Client<C>> {
+        Self::new_context_arc(config, native_config, rd_kafka_type, Arc::new(context))
+    }
+
+    /// Creates a new `Client` given a configuration, a client type and a context.
+    pub fn new_context_arc(
+        config: &ClientConfig,
+        native_config: NativeClientConfig,
+        rd_kafka_type: RDKafkaType,
+        context: Arc<C>,
     ) -> KafkaResult<Client<C>> {
         let mut err_buf = ErrBuf::new();
-        let context = Arc::new(context);
         unsafe {
             rdsys::rd_kafka_conf_set_opaque(
                 native_config.ptr(),

src/producer/base_producer.rs

@@ -258,10 +258,11 @@ where
         context: C,
     ) -> KafkaResult<BaseProducer<C, Part>> {
         let native_config = config.create_native_config()?;
+        let context = Arc::new(context);
 
         let partitioner = match context.get_custom_partitioner() {
             None => null_mut(),
-            Some(partitioner) => partitioner.as_ref() as *const Part as *mut c_void,
+            Some(partitioner) => partitioner as *const Part as *mut c_void,
         };
 
         if !partitioner.is_null() {
@@ -279,7 +280,7 @@ where
         unsafe {
             rdsys::rd_kafka_conf_set_dr_msg_cb(native_config.ptr(), Some(delivery_cb::<Part, C>))
         };
-        let client = Client::new(
+        let client = Client::new_context_arc(
             config,
             native_config,
             RDKafkaType::RD_KAFKA_PRODUCER,

src/producer/mod.rs

@@ -201,9 +201,7 @@ pub trait ProducerContext<Part: Partitioner = NoCustomPartitioner>: ClientContex
     fn delivery(&self, delivery_result: &DeliveryResult<'_>, delivery_opaque: Self::DeliveryOpaque);
 
     /// This method is called when creating producer in order to register custom partitioner.
-    /// Box is used to make sure data is on the heap as partitioner address will fly across FFI boundary.
-    #[allow(clippy::borrowed-box)]
-    fn get_custom_partitioner(&self) -> Option<&Box<Part>> {
+    fn get_custom_partitioner(&self) -> Option<&Part> {
         None
     }
 }

tests/test_low_producers.rs

@@ -94,7 +94,7 @@ impl<Part: Partitioner + Send + Sync> ProducerContext<Part> for CollectingContex
         }
     }
 
-    fn get_custom_partitioner(&self) -> Option<&Box<Part>> {
+    fn get_custom_partitioner(&self) -> Option<&Part> {
         match &self.partitioner {
             None => None,
             Some(p) => Some(&p),