@Override
    public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
        int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
    ) {
        log.warn("Broker " + brokerId + " does not support MAX_TIMESTAMP offset specs");
        Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
        for (TopicPartition topicPartition : keys) {
            Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
            if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
                maxTimestampPartitions.put(topicPartition, exception);
            }
        }
        // If there are no partitions with MAX_TIMESTAMP specs the UnsupportedVersionException cannot be handled
        // and all partitions should be failed here.
        // Otherwise, just the partitions with MAX_TIMESTAMP specs should be failed here and the fulfillment stage
        // will later be retried for the potentially empty set of partitions with non-MAX_TIMESTAMP specs.
        if (maxTimestampPartitions.isEmpty()) {
            return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
        } else {
            return maxTimestampPartitions;
        }
    }
@Test
    public void testHandleResponseUnsupportedVersion() {
        int brokerId = 1;
        UnsupportedVersionException uve = new UnsupportedVersionException("");
        Map<TopicPartition, OffsetSpec> maxTimestampPartitions = new HashMap<>();
        maxTimestampPartitions.put(t1p1, OffsetSpec.maxTimestamp());

        ListOffsetsHandler handler =
            new ListOffsetsHandler(offsetTimestampsByPartition, new ListOffsetsOptions(), logContext);

        final Map<TopicPartition, Long> nonMaxTimestampPartitions = new HashMap<>(offsetTimestampsByPartition);
        maxTimestampPartitions.forEach((k, v) -> nonMaxTimestampPartitions.remove(k));
        // Unsupported version exceptions currently cannot be handled if there's no partition with a
        // MAX_TIMESTAMP spec...
        Set<TopicPartition> keysToTest = nonMaxTimestampPartitions.keySet();
        Set<TopicPartition> expectedFailures = keysToTest;
        assertEquals(
            mapToError(expectedFailures, uve),
            handler.handleUnsupportedVersionException(brokerId, uve, keysToTest));

        // ...or if there are only partitions with MAX_TIMESTAMP specs.
        keysToTest = maxTimestampPartitions.keySet();
        expectedFailures = keysToTest;
        assertEquals(
            mapToError(expectedFailures, uve),
            handler.handleUnsupportedVersionException(brokerId, uve, keysToTest));

        // What can be handled is a request with a mix of partitions with MAX_TIMESTAMP specs
        // and partitions with non-MAX_TIMESTAMP specs.
        keysToTest = offsetTimestampsByPartition.keySet();
        expectedFailures = maxTimestampPartitions.keySet();
        assertEquals(
            mapToError(expectedFailures, uve),
            handler.handleUnsupportedVersionException(brokerId, uve, keysToTest));
    }