fn kafka<G>(
collection: Collection<G, (Option<Row>, Option<Row>), Diff>,
id: GlobalId,
connection: KafkaSinkConnection,
envelope: Option<SinkEnvelope>,
as_of: SinkAsOf,
write_frontier: Rc<RefCell<Antichain<Timestamp>>>,
metrics: &KafkaBaseMetrics,
connection_context: &ConnectionContext
) -> Rc<dyn Any> where
G: Scope<Timestamp = Timestamp>, 