| 250 | }); |
| 251 | |
| 252 | async fn make_test_operator( |
| 253 | shard: ShardId, |
| 254 | as_of: Antichain<Timestamp>, |
| 255 | ) -> ( |
| 256 | ReclockOperator< |
| 257 | kafka::KafkaTimestamp, |
| 258 | Timestamp, |
| 259 | impl RemapHandle<FromTime = kafka::KafkaTimestamp, IntoTime = Timestamp>, |
| 260 | >, |
| 261 | ReclockBatch<kafka::KafkaTimestamp, Timestamp>, |
| 262 | ) { |
| 263 | let metadata = CollectionMetadata { |
| 264 | persist_location: PersistLocation { |
| 265 | blob_uri: SensitiveUrl::from_str("mem://").expect("invalid URL"), |
| 266 | consensus_uri: SensitiveUrl::from_str("mem://").expect("invalid URL"), |
| 267 | }, |
| 268 | data_shard: shard, |
| 269 | relation_desc: RelationDesc::empty(), |
| 270 | txns_shard: None, |
| 271 | }; |
| 272 | |
| 273 | let write_frontier = Rc::new(RefCell::new(Antichain::from_elem(Timestamp::minimum()))); |
| 274 | |
| 275 | // Always in read-write mode for tests. |
| 276 | let (_read_only_tx, read_only_rx) = watch::channel(false); |
| 277 | let remap_handle = crate::source::reclock::compat::PersistHandle::new( |
| 278 | Arc::clone(&*PERSIST_CACHE), |
| 279 | read_only_rx, |
| 280 | metadata, |
| 281 | as_of.clone(), |
| 282 | write_frontier, |
| 283 | GlobalId::Explain, |
| 284 | "unittest", |
| 285 | 0, |
| 286 | 1, |
| 287 | PROGRESS_DESC.clone(), |
| 288 | GlobalId::Explain, |
| 289 | ) |
| 290 | .await |
| 291 | .unwrap(); |
| 292 | |
| 293 | let (mut operator, mut initial_batch) = ReclockOperator::new(remap_handle).await; |
| 294 | |
| 295 | // Push any updates that might already exist in the persist shard to the follower. |
| 296 | if *initial_batch.upper == [Timestamp::minimum()] { |
| 297 | // In the tests we always reclock the minimum source frontier to the minimum target |
| 298 | // frontier, which we do in this step. |
| 299 | initial_batch = operator |
| 300 | .mint( |
| 301 | 0.into(), |
| 302 | Antichain::from_elem(1.into()), |
| 303 | Antichain::from_elem(Partitioned::minimum()).borrow(), |
| 304 | ) |
| 305 | .await; |
| 306 | } |
| 307 | |
| 308 | (operator, initial_batch) |
| 309 | } |