code: def test_kafka_group_io_dataset_stream_timeout_check():
      """Test the functionality of the KafkaGroupIODataset when the
      consumer is configured to have a valid stream_timeout value and thus waits
      for the new messages from kafka.
      NOTE: The default value for message_timeout=5000
      """
      import tensorflow_io.kafka as kafka_io
  
      def write_messages_background():
          # Write new messages to the topic in a background thread
          time.sleep(6)
          for i in range(100, 200):
              message = f"D{i}"
              kafka_io.write_kafka(
                  message=message, topic="key-partition-test"
              )
  
      dataset = tfio.experimental.streaming.KafkaGroupIODataset(
          topics=["key-partition-test"],
          group_id="cgteststreamvalid",
          servers="localhost:9092",
          stream_timeout=20000,
          configuration=[
              "session.timeout.ms=7000",
              "max.poll.interval.ms=8000",
              "auto.offset.reset=earliest",
          ],
      )
  
      # start writing the new messages to kafka using the background job.
      # the job sleeps for some time (< stream_timeout) and then writes the
      # messages into the topic.
      thread = threading.Thread(
          target=write_messages_background, args=()
      )
      thread.daemon = True
      thread.start()
  
      # At the end, after the timeout has occurred, we must have the old 100 messages
      # along with the new 100 messages
      assert np.all(
          sorted(k.numpy() for (k, _) in dataset)
          == sorted(("D" + str(i)).encode() for i in range(200))
      )
  
imports: import sys
  import time
  import pytest
  import numpy
  import threading
  import tensorflow
  import tensorflow_io
label_desc: Use image node_id 16 for calling a global function with example usage: test_kafka_group_io_dataset_stream_timeout_check() without return types