rptest: Test for future offset for leader epoch

- Ensure that the returned epoch end offset is -1 for requested epoch's
that are in the future.

- Ensure that the returned leader epoch is not the current epoch but
actually the requested epoch.

(cherry picked from commit 1d5d6de)

tests/rptest/tests/offset_for_leader_epoch_test.py

@@ -108,11 +108,11 @@ def test_offset_for_leader_epoch(self):
 
  for o in leader_epoch_offsets:
  # check if the offset epoch matches what is expected or it is not available
- # (may be the case if leader wasn't elected in term 1 but other term in this case the offset for term 1 will not be presetn)
+ # (may be the case if leader wasn't elected in term 1 but other term in this case the offset for term 1 will not be present)
  assert initial_offsets[(o.topic,
  o.partition)] == o.epoch_end_offset or (
  o.epoch_end_offset == -1
- and o.leader_epoch > 1)
+ and o.leader_epoch == 1)
 
  # restart all the nodes to force leader election,
  # increase start timeout as partition count may get large
@@ -129,7 +129,7 @@ def test_offset_for_leader_epoch(self):
  assert initial_offsets[(o.topic,
  o.partition)] == o.epoch_end_offset or (
  o.epoch_end_offset == -1
- and o.leader_epoch > 2)
+ and o.leader_epoch == 1)
 
  last_offsets = self.list_offsets(topics=topics,
  total_partitions=total_partitions)
@@ -152,6 +152,16 @@ def test_offset_for_leader_epoch(self):
  for o in leader_epoch_offsets:
  assert o.error is not None and "UNKNOWN_LEADER_EPOCH" in o.error
 
+ # test case for requested_epoch larger then leader_epoch
+
+ leader_epoch_offsets = kcl.offset_for_leader_epoch(topics=topic_names,
+ leader_epoch=15000)
+
+ for o in leader_epoch_offsets:
+ # Ensure the leader_epoch returned is not the current leader_epoch
+ # but the requested
+ assert o.error == '' and o.leader_epoch == 15000 and o.epoch_end_offset == -1
+
  @cluster(num_nodes=6, log_allow_list=RESTART_LOG_ALLOW_LIST)
  def test_offset_for_leader_epoch_transfer(self):