code: def test_dataset_with_variable_sequence_length_truncates(self):
      spec = tf.TensorSpec((), tf.int64)
      table_spec = tf.TensorSpec((None,), tf.int64)
      table = reverb.Table(
          name=self._table_name,
          sampler=reverb.selectors.Fifo(),
          remover=reverb.selectors.Fifo(),
          max_times_sampled=1,
          max_size=100,
          rate_limiter=reverb.rate_limiters.MinSize(1),
          signature=table_spec,
      )
      server = reverb.Server([table])
      py_client = reverb.Client("localhost:{}".format(server.port))
  
      # Insert two episodes: one of length 3 and one of length 5
      with py_client.trajectory_writer(10) as writer:
          writer.append(1)
          writer.append(2)
          writer.append(3)
          writer.create_item(
              self._table_name,
              trajectory=writer.history[-3:],
              priority=5,
          )
  
      with py_client.trajectory_writer(10) as writer:
          writer.append(10)
          writer.append(20)
          writer.append(30)
          writer.append(40)
          writer.append(50)
          writer.create_item(
              self._table_name,
              trajectory=writer.history[-5:],
              priority=5,
          )
  
      replay = reverb_replay_buffer.ReverbReplayBuffer(
          spec,
          self._table_name,
          local_server=server,
          sequence_length=None,
          rate_limiter_timeout_ms=100,
      )
      ds = replay.as_dataset(
          single_deterministic_pass=True, num_steps=2
      )
      it = iter(ds)
  
      # Expect [1, 2]
      data, _ = next(it)
      self.assertAllEqual(data, [1, 2])
  
      # Expect [10, 20]
      data, _ = next(it)
      self.assertAllEqual(data, [10, 20])
  
      # Expect [30, 40]
      data, _ = next(it)
      self.assertAllEqual(data, [30, 40])
  
      with self.assertRaises(StopIteration):
          next(it)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from absl.testing import parameterized
  import reverb
  import tensorflow
  from tf_agents.drivers import py_driver
  from tf_agents.environments import test_envs
  from tf_agents.policies import random_py_policy
  from tf_agents.replay_buffers import reverb_replay_buffer
  from tf_agents.replay_buffers import reverb_utils
  from tf_agents.specs import array_spec
  from tf_agents.specs import tensor_spec
  from tf_agents.trajectories import trajectory
  from tf_agents.utils import common
  from tf_agents.utils import test_utils
label_desc: Use image node_id 5 for calling the ReverbReplayBufferTest obj's underlying member method code with example usage: obj.test_dataset_with_variable_sequence_length_truncates() without return types