bug: s3_v2 connector cannot ingest new add files

[tabVersion]

Describe the bug

originally posted by a user from the community https://risingwave-community.slack.com/archives/C03BW71523T/p1720701467609019

@wcy-fdu and I have identified the root cause that list exec just list the files once and does not keep looping, which causes the bug.

risingwave/src/stream/src/executor/source/list_executor.rs

Lines 82 to 106 in 1c65460

	let chunked_stream = stream.chunks(CHUNK_SIZE).map(|chunk| {
	let rows = chunk
	.into_iter()
	.map(|item| match item {
	Ok(page_item) => Ok((
	Op::Insert,
	OwnedRow::new(vec![
	Some(ScalarImpl::Utf8(page_item.name.into_boxed_str())),
	Some(ScalarImpl::Timestamptz(page_item.timestamp)),
	Some(ScalarImpl::Int64(page_item.size)),
	]),
	)),
	Err(e) => {
	tracing::error!(error = %e.as_report(), "Connector fail to list item");
	Err(e)
	}
	})
	.collect::<Vec<_>>();
	let res: Vec<(Op, OwnedRow)> = rows.into_iter().flatten().collect();
	Ok(StreamChunk::from_rows(
	&res,
	&[DataType::Varchar, DataType::Timestamptz, DataType::Int64],
	))
	});

Error message/log

No response

To Reproduce

No response

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response

[st1page]

Maybe there should be some test for this case 🤔

[marceloneppel]

Using the RisingWave operator, is it safe to reschedule the compute node pod (by, for example, killing the current pod) to force the ingestion of the newly added files?