Function mz_storage::sink::kafka::sink_collection

fn sink_collection<G: Scope<Timestamp = Timestamp>>(
    name: String,
    input: &Collection<G, KafkaMessage, Diff>,
    sink_id: GlobalId,
    connection: KafkaSinkConnection,
    partition_strategy: SinkPartitionStrategy,
    storage_configuration: StorageConfiguration,
    sink: &StorageSinkDesc<MetadataFilled, Timestamp>,
    metrics: KafkaSinkMetrics,
    statistics: SinkStatistics,
    write_frontier: Rc<RefCell<Antichain<Timestamp>>>,
) -> (Stream<G, HealthStatusMessage>, PressOnDropButton)

Sinks a collection of encoded rows to Kafka.

This operator exchanges all updates to a single worker by hashing on the given sink id.

Updates are sent in ascending timestamp order.