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

Code before:
        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(())
                    }
                    .boxed(),
                )
                .await?;
        }
        Ok(())
    }

    async fn barrier(&mut self, is_checkpoint: bool) -> risingwave_connector::sink::Result<()> {
        if is_checkpoint {
            self.store.inc_checkpoint();
        }
        Ok(())
    }
}

impl Drop for AsyncTruncateTestWriter {
    fn drop(&mut self) {
        self.parallelism_counter.fetch_sub(1, Relaxed);
    }
}

pub fn simple_name_of_id(id: i32) -> String {
    format!("name-{}", id)
}

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::commit`).
```suggestion
        _epoch: u64,
```

Code after:
        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,
        commit_metadata: Vec<u8>,
    ) -> risingwave_connector::sink::Result<()> {
        if commit_metadata.is_empty() {
            return Ok(());
        }

        if SimulationTestSink::random_err(&self.err_rate) {
            println!("commit with err");
            self.store.inc_err();
            return Err(SinkError::Internal(anyhow::anyhow!("fail to commit")));
        }

        let write_results_bytes: Vec<Vec<u8>> = serde_json::from_slice(&commit_metadata).unwrap();

        let file_ids = write_results_bytes
            .into_iter()
            .map(|metadata_bytes: Vec<u8>| {
                usize::from_le_bytes((metadata_bytes.as_slice()).try_into().unwrap())
            });

        self.store.insert_many(
            file_ids
                .into_iter()
                .map(|file_id| self.staging_store.get(file_id))
                .flatten(),