import tensorflow as tf

        coord = tf.train.Coordinator()
        tf.train.start_queue_runners(coord=coord)
        # Sleep to make sure the queue runner has started the first run call.
        time.sleep(_SLEEP_TIME)

      # Session closed.
      coord.request_stop()
      coord.join()

  def test_batcher_closed(self):
    with tf.Graph().as_default():
      @dynamic_batching.batch_fn
      def f(a):
        return a

      f(tf.constant([1]))

      # Intentionally using tf.Session() instead of self.test_session() to have
      # control over closing the session. test_session() is a cached session.
      with tf.Session():
        coord = tf.train.Coordinator()
        tf.train.start_queue_runners(coord=coord)
        time.sleep(_SLEEP_TIME)
        coord.request_stop()  # Calls close operation.
        coord.join()
      # Session closed.

  def test_minimum_batch_size(self):
    with self.test_session() as session:
      @dynamic_batching.batch_fn_with_options(
