public static Predicate<? super MetricKeyable> getSelectorFromRequestedMetrics(List<String> requestedMetrics) {
        if (requestedMetrics == null || requestedMetrics.isEmpty()) {
            log.debug("Telemetry subscription has specified no metric names; telemetry will record no metrics");
            return SELECTOR_NO_METRICS;
        } else if (requestedMetrics.size() == 1 && requestedMetrics.get(0) != null && requestedMetrics.get(0).equals("*")) {
            log.debug("Telemetry subscription has specified a single '*' metric name; using all metrics");
            return SELECTOR_ALL_METRICS;
        } else {
            log.debug("Telemetry subscription has specified to include only metrics that are prefixed with the following strings: {}", requestedMetrics);
            return k -> requestedMetrics.stream().anyMatch(f -> k.key().name().startsWith(f));
        }
    }
Examine the response data and handle different error code accordingly:

<ul>
    <li>Invalid Request: Disable Telemetry</li>
    <li>Invalid Record: Disable Telemetry</li>
    <li>Unsupported Version: Disable Telemetry</li>
    <li>UnknownSubscription or Unsupported Compression: Retry immediately</li>
    <li>TelemetryTooLarge or ThrottlingQuotaExceeded: Retry as per next interval</li>
</ul>

@param errorCode response body error code
@param intervalMs current push interval in milliseconds

@return Optional of push interval in milliseconds
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java