class_docstr: Tests for KafkaDataset.
docstr: Tests for KafkaDataset when reading non-keyed messages
  from a single-partitioned topic
docstr_tok: ["Tests","for","KafkaDataset","when","reading","non-keyed","messages","from","a","single-partitioned","topic"]
code: def test_kafka_dataset(self):
      """Tests for KafkaDataset when reading non-keyed messages
      from a single-partitioned topic"""
      tf.compat.v1.disable_eager_execution()
  
      import tensorflow_io.kafka as kafka_io  # pylint: disable=wrong-import-position
  
      topics = tf.compat.v1.placeholder(tf.dtypes.string, shape=[None])
      num_epochs = tf.compat.v1.placeholder(tf.dtypes.int64, shape=[])
      batch_size = tf.compat.v1.placeholder(tf.dtypes.int64, shape=[])
  
      repeat_dataset = kafka_io.KafkaDataset(
          topics, group="test", eof=True
      ).repeat(num_epochs)
      batch_dataset = repeat_dataset.batch(batch_size)
  
      iterator = tf.compat.v1.data.Iterator.from_structure(
          batch_dataset.output_types
      )
      init_op = iterator.make_initializer(repeat_dataset)
      init_batch_op = iterator.make_initializer(batch_dataset)
      get_next = iterator.get_next()
  
      with self.cached_session() as sess:
          # Basic test: read a limited number of messages from the topic.
          sess.run(
              init_op, feed_dict={topics: ["test:0:0:4"], num_epochs: 1}
          )
          for i in range(5):
              self.assertEqual(
                  ("D" + str(i)).encode(), sess.run(get_next)
              )
          with self.assertRaises(tf.errors.OutOfRangeError):
              sess.run(get_next)
  
          # Basic test: read all the messages from the topic from offset 5.
          sess.run(
              init_op,
              feed_dict={topics: ["test:0:5:-1"], num_epochs: 1},
          )
          for i in range(5):
              self.assertEqual(
                  ("D" + str(i + 5)).encode(), sess.run(get_next)
              )
          with self.assertRaises(tf.errors.OutOfRangeError):
              sess.run(get_next)
  
          # Basic test: read from different subscriptions of the same topic.
          sess.run(
              init_op,
              feed_dict={
                  topics: ["test:0:0:4", "test:0:5:-1"],
                  num_epochs: 1,
              },
          )
          for j in range(2):
              for i in range(5):
                  self.assertEqual(
                      ("D" + str(i + j * 5)).encode(),
                      sess.run(get_next),
                  )
          with self.assertRaises(tf.errors.OutOfRangeError):
              sess.run(get_next)
  
          # Test repeated iteration through both subscriptions.
          sess.run(
              init_op,
              feed_dict={
                  topics: ["test:0:0:4", "test:0:5:-1"],
                  num_epochs: 10,
              },
          )
          for _ in range(10):
              for j in range(2):
                  for i in range(5):
                      self.assertEqual(
                          ("D" + str(i + j * 5)).encode(),
                          sess.run(get_next),
                      )
          with self.assertRaises(tf.errors.OutOfRangeError):
              sess.run(get_next)
  
          # Test batched and repeated iteration through both subscriptions.
          sess.run(
              init_batch_op,
              feed_dict={
                  topics: ["test:0:0:4", "test:0:5:-1"],
                  num_epochs: 10,
                  batch_size: 5,
              },
          )
          for _ in range(10):
              self.assertAllEqual(
                  [("D" + str(i)).encode() for i in range(5)],
                  sess.run(get_next),
              )
              self.assertAllEqual(
                  [("D" + str(i + 5)).encode() for i in range(5)],
                  sess.run(get_next),
              )
  
file_path: io/tests/test_kafka_v1.py
imports: import time
  import pytest
  import tensorflow
label_desc: Use image node_id 1 for calling the KafkaDatasetTest obj's underlying member method code with example usage: obj.test_kafka_dataset() without return types