class_docstr: Tests for KafkaDataset.
docstr: test_write_kafka
docstr_tok: ["test_write_kafka"]
code: def test_write_kafka(self):
      """test_write_kafka"""
      tf.compat.v1.disable_eager_execution()
  
      import tensorflow_io.kafka as kafka_io  # pylint: disable=wrong-import-position
  
      channel = f"e{time.time()}e"
  
      # Start with reading test topic, replace `D` with `e(time)e`,
      # and write to test_e(time)e` topic.
      dataset = kafka_io.KafkaDataset(
          topics=["test:0:0:4"], group="test", eof=True
      )
      dataset = dataset.map(
          lambda x: kafka_io.write_kafka(
              tf.strings.regex_replace(x, "D", channel),
              topic="test_" + channel,
          )
      )
      iterator = dataset.make_initializable_iterator()
      init_op = iterator.initializer
      get_next = iterator.get_next()
  
      with self.cached_session() as sess:
          # Basic test: read from topic 0.
          sess.run(init_op)
          for i in range(5):
              self.assertEqual(
                  (channel + str(i)).encode(), sess.run(get_next)
              )
          with self.assertRaises(tf.errors.OutOfRangeError):
              sess.run(get_next)
  
      # Reading from `test_e(time)e` we should get the same result
      dataset = kafka_io.KafkaDataset(
          topics=["test_" + channel], group="test", eof=True
      )
      iterator = dataset.make_initializable_iterator()
      init_op = iterator.initializer
      get_next = iterator.get_next()
  
      with self.cached_session() as sess:
          sess.run(init_op)
          for i in range(5):
              self.assertEqual(
                  (channel + str(i)).encode(), sess.run(get_next)
              )
          with self.assertRaises(tf.errors.OutOfRangeError):
              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 7 for calling the KafkaDatasetTest obj's underlying member method code with example usage: obj.test_write_kafka() without return types