public static ApiVersionCollection intersectForwardableApis(
        final ApiMessageType.ListenerType listenerType,
        final RecordVersion minRecordVersion,
        final Map<ApiKeys, ApiVersion> activeControllerApiVersions,
        boolean enableUnstableLastVersion,
        boolean clientTelemetryEnabled
    ) {
        ApiVersionCollection apiKeys = new ApiVersionCollection();
        for (ApiKeys apiKey : ApiKeys.apisForListener(listenerType)) {
            if (apiKey.minRequiredInterBrokerMagic <= minRecordVersion.value) {
                final Optional<ApiVersion> brokerApiVersion = apiKey.toApiVersion(enableUnstableLastVersion);
                if (!brokerApiVersion.isPresent()) {
                    // Broker does not support this API key.
                    continue;
                }

                // Skip telemetry APIs if client telemetry is disabled.
                if ((apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey == ApiKeys.PUSH_TELEMETRY) && !clientTelemetryEnabled)
                    continue;

                final ApiVersion finalApiVersion;
                if (!apiKey.forwardable) {
                    finalApiVersion = brokerApiVersion.get();
                } else {
                    Optional<ApiVersion> intersectVersion = intersect(
                        brokerApiVersion.get(),
                        activeControllerApiVersions.getOrDefault(apiKey, null)
                    );
                    if (intersectVersion.isPresent()) {
                        finalApiVersion = intersectVersion.get();
                    } else {
                        // Controller doesn't support this API key, or there is no intersection.
                        continue;
                    }
                }

                apiKeys.add(finalApiVersion.duplicate());
            }
        }
        return apiKeys;
    }
Find the common range of supported API versions between the locally
known range and that of another set.

@param listenerType the listener type which constrains the set of exposed APIs
@param minRecordVersion min inter broker magic
@param activeControllerApiVersions controller ApiVersions
@param enableUnstableLastVersion whether unstable versions should be advertised or not
@param clientTelemetryEnabled whether client telemetry is enabled or not
@return commonly agreed ApiVersion collection
clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java