Repository: risingwavelabs/risingwave
File: src/meta/src/manager/sink_coordination/coordinator_worker.rs (Rust)

Code before:
                        epoch,
                        handle_id
                    );
                }
                CoordinationHandleManagerEvent::AlignInitialEpoch(epoch) => {
                    bail!(
                        "receive AlignInitialEpoch on epoch {} from handle {} during alter parallelism",
                        epoch,
                        handle_id
                    );
                }
            }
        }
        Ok(requests.handle_ids)
    }
}

enum CoordinatorWorkerState {
    Running,
    WaitingForFlushed(HashSet<HandleId>),
}

pub struct CoordinatorWorker {
    handle_manager: CoordinationHandleManager,
    prev_committed_epoch: Option<u64>,
    curr_state: CoordinatorWorkerState,
}

enum CoordinatorWorkerEvent {
    HandleManagerEvent(HandleId, CoordinationHandleManagerEvent),
    ReadyToCommit(u64, Vec<u8>),
}

impl CoordinatorWorker {
    pub async fn run(
        param: SinkParam,
        request_rx: UnboundedReceiver<SinkWriterCoordinationHandle>,
        db: DatabaseConnection,
        subscriber: SinkCommittedEpochSubscriber,
        iceberg_compact_stat_sender: UnboundedSender<IcebergSinkCompactionUpdate>,
    ) {
        let sink = match build_sink(param.clone()) {
            Ok(sink) => sink,
            Err(e) => {
                error!(
                    error = %e.as_report(),
                    "unable to build sink with param {:?}",
                    param
                );
                return;
            }

Review comment (suggestion):
[nitpick] The `CoordinatorWorkerState` enum lacks documentation explaining its purpose and the state transitions. Consider adding a doc comment like:

```rust
/// Represents the coordinator worker's state machine for handling schema changes.
/// 
/// - `Running`: Normal operation, handles can be started immediately
/// - `WaitingForFlushed`: Waiting for all pending two-phase commits to complete before
///   starting new handles. This ensures new sink executors load the correct schema.
enum CoordinatorWorkerState {
```

This would help developers understand the synchronization mechanism described in the PR.

Code after:
                        epoch,
                        handle_id
                    );
                }
                CoordinationHandleManagerEvent::AlignInitialEpoch(epoch) => {
                    bail!(
                        "receive AlignInitialEpoch on epoch {} from handle {} during alter parallelism",
                        epoch,
                        handle_id
                    );
                }
            }
        }
        Ok(requests.handle_ids)
    }
}

/// Represents the coordinator worker's state machine for handling schema changes.
///
/// - `Running`: Normal operation, handles can be started immediately
/// - `WaitingForFlushed`: Waiting for all pending two-phase commits to complete before starting new handles. This
///   ensures new sink executors load the correct schema.
enum CoordinatorWorkerState {
    Running,
    WaitingForFlushed(HashSet<HandleId>),
}

pub struct CoordinatorWorker {
    handle_manager: CoordinationHandleManager,
    prev_committed_epoch: Option<u64>,
    curr_state: CoordinatorWorkerState,
}

enum CoordinatorWorkerEvent {
    HandleManagerEvent(HandleId, CoordinationHandleManagerEvent),
    ReadyToCommit(u64, Vec<u8>),
}

impl CoordinatorWorker {
    pub async fn run(
        param: SinkParam,
        request_rx: UnboundedReceiver<SinkWriterCoordinationHandle>,
        db: DatabaseConnection,
        subscriber: SinkCommittedEpochSubscriber,
        iceberg_compact_stat_sender: UnboundedSender<IcebergSinkCompactionUpdate>,
    ) {
        let sink = match build_sink(param.clone()) {
            Ok(sink) => sink,
            Err(e) => {
                error!(
                    error = %e.as_report(),