Function ensure_producer_offsets_table 

pub fn ensure_producer_offsets_table(
    conn: &Connection,
) -> Result<(), BookkeepingError>

Create the foreign-write producer watermark table.