fix ci

Author: Lori-Shu

rocketmq-broker/src/transaction/queue/default_transactional_message_service.rs

@@ -131,7 +131,7 @@ where
         let mut send_map = HashMap::<i32, Message>::new();
         let delete_context_mutex_guard = self.delete_context.lock().await;
         for (queue_id, mq_context) in delete_context_mutex_guard.iter() {
-            if mq_context.get_total_size().await <= 0
+            if mq_context.get_total_size().await == 0
                 || mq_context.is_empty().await
                 || (mq_context.get_total_size().await < max_size as u32
                     && (start_time as i64 - mq_context.get_last_write_timestamp().await as i64)

rocketmq-broker/src/transaction/queue/message_queue_op_context.rs

@@ -49,9 +49,7 @@ impl MessageQueueOpContext {
         if self.context_receiver.len() > self.queue_capacity {
             return Err(anyhow::Error::msg("queue is full".to_string()));
         }
-        self.context_queue
-            .send(msg)
-            .map_err(|e| anyhow::Error::new(e))
+        self.context_queue.send(msg).map_err(anyhow::Error::new)
     }
     pub async fn offer(&self, item: String, timeout: std::time::Duration) -> Result<()> {
         time::timeout(timeout, self.push(item)).await.unwrap()