@Override
    public LookupResult<BrokerKey> handleResponse(Set<BrokerKey> keys, AbstractResponse abstractResponse) {
        validateLookupKeys(keys);

        MetadataResponse response = (MetadataResponse) abstractResponse;
        MetadataResponseData.MetadataResponseBrokerCollection brokers = response.data().brokers();

        if (brokers.isEmpty()) {
            log.debug("Metadata response contained no brokers. Will backoff and retry");
            return LookupResult.empty();
        } else {
            log.debug("Discovered all brokers {} to send requests to", brokers);
        }

        Map<BrokerKey, Integer> brokerKeys = brokers.stream().collect(Collectors.toMap(
            broker -> new BrokerKey(OptionalInt.of(broker.nodeId())),
            MetadataResponseData.MetadataResponseBroker::nodeId
        ));

        return new LookupResult<>(
            Collections.singletonList(ANY_BROKER),
            Collections.emptyMap(),
            brokerKeys
        );
    }
@Test
    public void testHandleResponseWithInvalidLookupKeys() {
        AllBrokersStrategy strategy = new AllBrokersStrategy(logContext);
        AllBrokersStrategy.BrokerKey key1 = new AllBrokersStrategy.BrokerKey(OptionalInt.empty());
        AllBrokersStrategy.BrokerKey key2 = new AllBrokersStrategy.BrokerKey(OptionalInt.of(1));
        MetadataResponse response = new MetadataResponse(new MetadataResponseData(), ApiKeys.METADATA.latestVersion());

        assertThrows(IllegalArgumentException.class, () -> strategy.handleResponse(mkSet(key1), response));
        assertThrows(IllegalArgumentException.class, () -> strategy.handleResponse(mkSet(key2), response));
        assertThrows(IllegalArgumentException.class, () -> strategy.handleResponse(mkSet(key1, key2), response));

        Set<AllBrokersStrategy.BrokerKey> keys = new HashSet<>(AllBrokersStrategy.LOOKUP_KEYS);
        keys.add(key2);
        assertThrows(IllegalArgumentException.class, () -> strategy.handleResponse(keys, response));
    }