code: def test_from_kafka():
      j = random.randint(0, 10000)
      ARGS = {
          "bootstrap.servers": "localhost:9092",
          "group.id": "streamz-test%i" % j,
      }
      with kafka_service() as kafka:
          kafka, TOPIC = kafka
          stream = Stream.from_kafka([TOPIC], ARGS, asynchronous=True)
          out = stream.sink_to_list()
          stream.start()
          yield gen.sleep(1.1)  # for loop to run
          for i in range(10):
              yield gen.sleep(
                  0.1
              )  # small pause ensures correct ordering
              kafka.produce(TOPIC, b"value-%d" % i)
          kafka.flush()
          # it takes some time for messages to come back out of kafka
          wait_for(lambda: len(out) == 10, 10, period=0.1)
          assert out[-1] == b"value-9"
  
          kafka.produce(TOPIC, b"final message")
          kafka.flush()
          wait_for(lambda: out[-1] == b"final message", 10, period=0.1)
  
          stream._close_consumer()
          kafka.produce(TOPIC, b"lost message")
          kafka.flush()
          # absolute sleep here, since we expect output list *not* to change
          yield gen.sleep(1)
          assert out[-1] == b"final message"
          stream._close_consumer()
  
imports: import asyncio
  import atexit
  from contextlib import contextmanager
  from flaky import flaky
  import os
  import pytest
  import random
  import shlex
  import subprocess
  import time
  from tornado import gen
  from ..core import Stream
  from ..dask import DaskStream
  from streamz.utils_test import gen_test, wait_for, await_for
  from distributed.utils_test import gen_cluster
label_desc: Use image node_id 5 for calling a global function with example usage: test_from_kafka() without return types