pub(crate) async fn cleanup_dropped_streaming_jobs(
refresh_manager: &GlobalRefreshManagerRef,
hummock_manager: &HummockManagerRef,
metadata_manager: &MetadataManager,
sink_manager: &SinkCoordinatorManager,
iceberg_pk_index_sink_manager: &IcebergPkIndexSinkManager,
streaming_job_ids: impl IntoIterator<Item = JobId>,
state_table_ids: Vec<TableId>,
progress_status: &str,
) -> MetaResult<()>