@Override
    ListOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
        Map<String, ListOffsetsTopic> topicsByName = CollectionUtils.groupPartitionsByTopic(
            keys,
            topicName -> new ListOffsetsTopic().setName(topicName),
            (listOffsetsTopic, partitionId) -> {
                TopicPartition topicPartition = new TopicPartition(listOffsetsTopic.name(), partitionId);
                long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
                listOffsetsTopic.partitions().add(
                    new ListOffsetsPartition()
                        .setPartitionIndex(partitionId)
                        .setTimestamp(offsetTimestamp));
            });
        boolean supportsMaxTimestamp = keys
            .stream()
            .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP);

        boolean requireTieredStorageTimestamp = keys
            .stream()
            .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP || offsetTimestampsByPartition.get(key) == ListOffsetsRequest.LATEST_TIERED_TIMESTAMP);

        return ListOffsetsRequest.Builder
            .forConsumer(true, options.isolationLevel(), supportsMaxTimestamp, requireTieredStorageTimestamp)
            .setTargetTimes(new ArrayList<>(topicsByName.values()));
    }
@Test
    public void testBuildRequestAllowedVersions() {
        ListOffsetsHandler defaultOptionsHandler =
            new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);
        ListOffsetsRequest.Builder builder =
            defaultOptionsHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
        assertEquals(1, builder.oldestAllowedVersion());

        ListOffsetsHandler readCommittedHandler =
            new ListOffsetsHandler(
                offsetTimestampsByPartition, new ListOffsetsOptions(IsolationLevel.READ_COMMITTED), logContext);
        builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0));
        assertEquals(2, builder.oldestAllowedVersion());

        builder = readCommittedHandler.buildBatchedRequest(node.id(), mkSet(t0p0, t0p1, t1p0, t1p1));
        assertEquals(7, builder.oldestAllowedVersion());
    }