Repository: risingwavelabs/risingwave
File: src/tests/simulation/tests/integration_tests/sink/utils.rs (Rust)

Code before:
        });
        self.store.insert_many(
            file_ids
                .into_iter()
                .map(|file_id| self.staging_store.get(file_id))
                .flatten(),
        );
        Ok(())
    }
}

pub struct AsyncTruncateTestWriter {
    store: TestSinkStore,
    parallelism_counter: Arc<AtomicUsize>,
    err_rate: Arc<AtomicU32>,
}

impl AsyncTruncateSinkWriter for AsyncTruncateTestWriter {
    type DeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + Send + 'static;

    async fn write_chunk<'a>(
        &'a mut self,
        chunk: StreamChunk,
        mut add_future: DeliveryFutureManagerAddFuture<'a, Self::DeliveryFuture>,
    ) -> risingwave_connector::sink::Result<()> {
        if SimulationTestSink::random_err(&self.err_rate) {
            println!("write with err");
            self.store.inc_err();
            return Err(SinkError::Internal(anyhow::anyhow!("fail to write")));
        }
        for (op, row) in chunk.rows() {
            assert_eq!(op, Op::Insert);
            assert_eq!(row.len(), 2);
            let id = row.datum_at(0).unwrap().into_int32();
            let name = row.datum_at(1).unwrap().into_utf8().to_string();
            let store = self.store.clone();
            let ack_err = SimulationTestSink::random_err(&self.err_rate);
            add_future
                .add_future_may_await(
                    async move {
                        if thread_rng().random_bool(0.5) {
                            let micro = (thread_rng().random::<u32>() as f64 * 30.0
                                / (u32::MAX as f64)) as _;
                            sleep(Duration::from_micros(micro)).await;
                        } else if ack_err {
                            println!("ack with err");
                            store.inc_err();
                            return Err(SinkError::Internal(anyhow::anyhow!("fail to ack")));
                        }
                        store.insert(id, name);
                        Ok(())

Review comment (suggestion):
The `epoch` parameter is unused in this method. Consider prefixing it with an underscore (`_epoch`) to indicate it's intentionally unused, which is consistent with other implementations in the codebase (e.g., `IcebergSinkCommitter::abort`).
```suggestion
        _epoch: u64,
```

Code after:
            self.store.inc_err();
            return Err(SinkError::Internal(anyhow::anyhow!("fail to commit")));
        }
        let file_ids = metadata.into_iter().map(|metadata| {
            let Metadata::Serialized(serialized) = metadata.metadata.unwrap();
            usize::from_le_bytes((serialized.metadata.as_slice()).try_into().unwrap())
        });
        self.store.insert_many(
            file_ids
                .into_iter()
                .map(|file_id| self.staging_store.get(file_id))
                .flatten(),
        );
        Ok(())
    }
}

#[async_trait]
impl TwoPhaseCommitCoordinator for TestCoordinator {
    async fn init(&mut self) -> risingwave_connector::sink::Result<()> {
        Ok(())
    }

    async fn pre_commit(
        &mut self,
        epoch: u64,
        metadata: Vec<SinkMetadata>,
        _add_columns: Option<Vec<Field>>,
    ) -> risingwave_connector::sink::Result<Vec<u8>> {
        if SimulationTestSink::random_err(&self.err_rate) {
            println!("pre-commit with err");
            self.store.inc_err();
            return Err(SinkError::Internal(anyhow::anyhow!("fail to pre-commit")));
        }

        let mut pre_commit_metadata_bytes = Vec::new();
        for metadata in metadata {
            let Metadata::Serialized(serialized) = metadata.metadata.unwrap();

            pre_commit_metadata_bytes.push(serialized.metadata);
        }

        let pre_commit_metadata_bytes: Vec<u8> =
            serde_json::to_vec(&pre_commit_metadata_bytes).unwrap();

        Ok(pre_commit_metadata_bytes)
    }

    async fn commit(
        &mut self,
        epoch: u64,