Stream ingestion worker