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

Code before:
        }
        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)
}

pub struct SimulationTestSink {
    _sink_guard: TestSinkRegistryGuard,
    pub store: TestSinkStore,
    pub parallelism_counter: Arc<AtomicUsize>,
    pub err_rate: Arc<AtomicU32>,
}

pub enum TestSinkType {
    SinkWriter,
    CoordinatedSink,
    AsyncTruncate,
}

#[macro_export]
macro_rules! for_all_sink_types {
    ($macro:path) => {
        $macro! {
            SinkWriter,
            CoordinatedSink,
            AsyncTruncate,
        }
    };
}

impl SimulationTestSink {
    pub fn register_new(test_type: TestSinkType) -> Self {
        let parallelism_counter = Arc::new(AtomicUsize::new(0));
        let err_rate = Arc::new(AtomicU32::new(0));
        let store = TestSinkStore::new();

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
    async fn abort(&mut self, _epoch: u64, commit_metadata: Vec<u8>) {
```

Code after:
        }

        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(),
        );
        Ok(())
    }

    async fn abort(&mut self, epoch: u64, commit_metadata: Vec<u8>) {
        // No-op
    }
}

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);