Skip to content
This repository was archived by the owner on Jan 16, 2026. It is now read-only.
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions crates/derive/src/stages/channel_reader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -264,6 +264,15 @@ mod test {
data.into()
}

#[tokio::test]
async fn test_flush_channel_reader() {
let mock = MockChannelReaderProvider::new(vec![Ok(Some(new_compressed_batch_data()))]);
let mut reader = ChannelReader::new(mock, Arc::new(RollupConfig::default()));
reader.next_batch = Some(BatchReader::from(new_compressed_batch_data()));
reader.flush_channel().await.unwrap();
assert!(reader.next_batch.is_none());
}

#[tokio::test]
async fn test_next_batch_batch_reader_set_fails() {
let mock = MockChannelReaderProvider::new(vec![Err(PipelineError::Eof.temp())]);
Expand Down