fix(common): only warn on dropping non-empty builder (risingwavelabs#8494)

Signed-off-by: Bugen Zhao <[email protected]>

Author: BugenZhao

src/common/src/util/chunk_coalesce.rs

@@ -229,6 +229,18 @@ impl DataChunkBuilder {
     }
 }
 
+impl Drop for DataChunkBuilder {
+    fn drop(&mut self) {
+        // Possible to fail when async task gets cancelled.
+        if self.buffered_count != 0 {
+            tracing::warn!(
+                remaining = self.buffered_count,
+                "dropping non-empty data chunk builder"
+            );
+        }
+    }
+}
+
 /// The iterator that yields data chunks during appending a data chunk to a [`DataChunkBuilder`].
 pub struct AppendDataChunk<'a> {
     builder: &'a mut DataChunkBuilder,
@@ -249,6 +261,7 @@ impl FusedIterator for AppendDataChunk<'_> {}
 
 impl Drop for AppendDataChunk<'_> {
     fn drop(&mut self) {
+        // Possible to fail when async task gets cancelled.
         if self.remaining.is_some() {
             tracing::warn!("dropping `AppendDataChunk` without exhausting it");
         }

src/stream/src/common/builder.rs

@@ -45,8 +45,13 @@ pub struct StreamChunkBuilder {
 
 impl Drop for StreamChunkBuilder {
     fn drop(&mut self) {
-        // Possible to fail in some corner cases but should not in unit tests
-        debug_assert_eq!(self.size, 0, "dropping non-empty stream chunk builder");
+        // Possible to fail when async task gets cancelled.
+        if self.size != 0 {
+            tracing::warn!(
+                remaining = self.size,
+                "dropping non-empty stream chunk builder"
+            );
+        }
     }
 }