public static Optional<Integer> maybeFetchErrorIntervalMs(short errorCode, int intervalMs) {
        if (errorCode == Errors.NONE.code())
            return Optional.empty();

        int pushIntervalMs;
        String reason;

        Errors error = Errors.forCode(errorCode);
        switch (error) {
            case INVALID_REQUEST:
            case INVALID_RECORD:
            case UNSUPPORTED_VERSION:
                pushIntervalMs = Integer.MAX_VALUE;
                reason = "The broker response indicates the client sent an request that cannot be resolved"
                    + " by re-trying, hence disable telemetry";
                break;
            case UNKNOWN_SUBSCRIPTION_ID:
            case UNSUPPORTED_COMPRESSION_TYPE:
                pushIntervalMs = 0;
                reason = error.message();
                break;
            case TELEMETRY_TOO_LARGE:
            case THROTTLING_QUOTA_EXCEEDED:
                reason = error.message();
                pushIntervalMs = (intervalMs != -1) ? intervalMs : ClientTelemetryReporter.DEFAULT_PUSH_INTERVAL_MS;
                break;
            default:
                reason = "Unwrapped error code";
                log.error("Error code: {}. Unmapped error for telemetry, disable telemetry.", errorCode);
                pushIntervalMs = Integer.MAX_VALUE;
        }

        log.debug("Error code: {}, reason: {}. Push interval update to {} ms.", errorCode, reason, pushIntervalMs);
        return Optional.of(pushIntervalMs);
    }
@Test
    public void testMaybeFetchErrorIntervalMs() {
        assertEquals(Optional.empty(), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.NONE.code(), -1));
        assertEquals(Optional.of(Integer.MAX_VALUE), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.INVALID_REQUEST.code(), -1));
        assertEquals(Optional.of(Integer.MAX_VALUE), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.INVALID_RECORD.code(), -1));
        assertEquals(Optional.of(0), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.UNKNOWN_SUBSCRIPTION_ID.code(), -1));
        assertEquals(Optional.of(0), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.UNSUPPORTED_COMPRESSION_TYPE.code(), -1));
        assertEquals(Optional.of(ClientTelemetryReporter.DEFAULT_PUSH_INTERVAL_MS), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.TELEMETRY_TOO_LARGE.code(), -1));
        assertEquals(Optional.of(20000), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.TELEMETRY_TOO_LARGE.code(), 20000));
        assertEquals(Optional.of(ClientTelemetryReporter.DEFAULT_PUSH_INTERVAL_MS), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.THROTTLING_QUOTA_EXCEEDED.code(), -1));
        assertEquals(Optional.of(20000), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.THROTTLING_QUOTA_EXCEEDED.code(), 20000));
        assertEquals(Optional.of(Integer.MAX_VALUE), ClientTelemetryUtils.maybeFetchErrorIntervalMs(Errors.UNKNOWN_SERVER_ERROR.code(), -1));
    }