private Optional<HashedDataFeedKey> lookupAndValidateKey(final String unHashedKey,
                                                             final AttributeMap attributeMap,
                                                             final ReceiveDataConfig receiveDataConfig) {
        if (!looksLikeADatafeedKey(unHashedKey, attributeMap)) {
            return Optional.empty();
        }

        // If we get here we are dealing with a key that looks like a data feed key,
        // so we can validate it as such

        final CIKey keyOwnerMetaKey = getOwnerMetaKey(receiveDataConfig);
        final String keyOwner = NullSafe.get(
                attributeMap,
                map -> map.get(keyOwnerMetaKey.get()),
                String::trim);
        if (NullSafe.isBlankString(keyOwner)) {
            LOGGER.debug("Blank keyOwner, attributeMap: {}", attributeMap);
            throw new StroomStreamException(
                    StroomStatusCode.DATA_FEED_KEY_NOT_AUTHENTICATED,
                    attributeMap,
                    "Mandatory header '" + keyOwnerMetaKey +
                    "' must be provided to authenticate with a data feed key.");
        }

        // We cache already authenticated un-hashed keys to save the hashing cost.
        // Cache get will trigger the hashing and checking if not found
        final Set<CachedHashedDataFeedKey> dataFeedKeys = unHashedKeyToDataFeedKeyCache.get(
                new UnHashedCacheKey(unHashedKey, CIKey.ofDynamicKey(keyOwner)));

        if (NullSafe.isEmptyCollection(dataFeedKeys)) {
            LOGGER.debug("Unknown data feed key {}, attributeMap: {}", unHashedKey, attributeMap);
            // We need to throw here because we know it looks like a DFK, so
            // we don't want to pass it on to the next filter.
            throw new StroomStreamException(StroomStatusCode.DATA_FEED_KEY_NOT_AUTHENTICATED, attributeMap);
        }

        final Predicate<CachedHashedDataFeedKey> filter = createKeyOwnerFilter(keyOwnerMetaKey, keyOwner);

        // In the event that we have more than one key for an accountId, get the one with
        // the latest expire time.
        final List<HashedDataFeedKey> filteredKeys = dataFeedKeys.stream()
                .filter(filter)
                .sorted(HASHED_DATA_FEED_KEY_COMPARATOR)
                .map(CachedHashedDataFeedKey::getDataFeedKey)
                .toList();

        if (NullSafe.isEmptyCollection(filteredKeys)) {
            // Got a match on the un-hashed key, but is a bad accountId.
            // Stops someone using their own DFK to authenticate to someone else's accountId.
            LOGGER.debug("No un-expired keys matching data feed key {}, keyOwnerMetaKey: {}, " +
                         "keyOwner: {}, dataFeedKeys: {}",
                    unHashedKey, keyOwnerMetaKey, keyOwner, dataFeedKeys);
            if (DATA_FEED_KEY_PATTERN.matcher(unHashedKey).matches()) {
                // It looks like a DFK, but we go no match, so we can throw
                throw new StroomStreamException(StroomStatusCode.DATA_FEED_KEY_NOT_AUTHENTICATED, attributeMap);
            } else {
                // Doesn't look like a DFK so let the next filter have a go
                return Optional.empty();
            }
        } else {
            // Very, very unlikely to have >1 item, but you never know
            final Optional<HashedDataFeedKey> optValidKey = filteredKeys.stream()
                    .filter(Predicate.not(HashedDataFeedKey::isExpired))
                    .findFirst();

            if (optValidKey.isEmpty()) {
                throw new StroomStreamException(StroomStatusCode.DATA_FEED_KEY_EXPIRED, attributeMap);
            }
            return optValidKey;
        }
    }
@return A populated {@link Optional} if the key is known to use and is valid, else empty.
stroom-receive/stroom-receive-common/src/main/java/stroom/receive/common/DataFeedKeyServiceImpl.java