Use CStr::to_string_lossy in Base{Consumer,Producer}

In some error cases, the `Base{Consumer,Producer}` were eagerly copying strings,
and `unwrap`ing utf8 validation, just to print an error message.

This will avoid the allocation in the common case, and be panic-safe in the presumably unreachable case of invalid utf-8.

Author: Swatinem

src/consumer/base_consumer.rs

@@ -149,12 +149,11 @@ where
                         self.handle_offset_commit_event(event);
                     }
                     _ => {
-                        let buf = unsafe {
+                        let evname = unsafe {
                             let evname = rdsys::rd_kafka_event_name(event.ptr());
-                            CStr::from_ptr(evname).to_bytes()
+                            CStr::from_ptr(evname).to_string_lossy()
                         };
-                        let evname = String::from_utf8(buf.to_vec()).unwrap();
-                        warn!("Ignored event '{}' on consumer poll", evname);
+                        warn!("Ignored event '{evname}' on consumer poll");
                     }
                 }
             }
@@ -192,13 +191,12 @@ where
                     .rebalance(self.client.native_client(), err, &mut tpl);
             }
             _ => {
-                let buf = unsafe {
+                let err = unsafe {
                     let err_name =
                         rdsys::rd_kafka_err2name(rdsys::rd_kafka_event_error(event.ptr()));
-                    CStr::from_ptr(err_name).to_bytes()
+                    CStr::from_ptr(err_name).to_string_lossy()
                 };
-                let err = String::from_utf8(buf.to_vec()).unwrap();
-                warn!("invalid rebalance event: {:?}", err);
+                warn!("invalid rebalance event: {err}");
             }
         }
     }

src/mocking.rs

@@ -137,7 +137,7 @@ where
     pub fn bootstrap_servers(&self) -> String {
         let bootstrap =
             unsafe { CStr::from_ptr(rdsys::rd_kafka_mock_cluster_bootstraps(self.mock_cluster)) };
-        bootstrap.to_string_lossy().to_string()
+        bootstrap.to_string_lossy().into_owned()
     }
 
     /// Clear the cluster's error state for the given ApiKey.

src/producer/base_producer.rs

@@ -368,11 +368,10 @@ where
             match evtype {
                 rdsys::RD_KAFKA_EVENT_DR => self.handle_delivery_report_event(ev),
                 _ => {
-                    let buf = unsafe {
+                    let evname = unsafe {
                         let evname = rdsys::rd_kafka_event_name(ev.ptr());
-                        CStr::from_ptr(evname).to_bytes()
+                        CStr::from_ptr(evname).to_string_lossy()
                     };
-                    let evname = String::from_utf8(buf.to_vec()).unwrap();
                     warn!("Ignored event '{}' on base producer poll", evname);
                 }
             }