public GetTelemetrySubscriptionsResponse processGetTelemetrySubscriptionRequest(
        GetTelemetrySubscriptionsRequest request, RequestContext requestContext) {

        long now = time.milliseconds();
        Uuid clientInstanceId = Optional.ofNullable(request.data().clientInstanceId())
            .filter(id -> !id.equals(Uuid.ZERO_UUID))
            .orElse(generateNewClientId());

        /*
         Get the client instance from the cache or create a new one. If subscription has changed
         since the last request, then the client instance will be re-evaluated. Validation of the
         request will be done after the client instance is created. If client issues another get
         telemetry request prior to push interval, then the client should get a throttle error but if
         the subscription has changed since the last request then the client should get the updated
         subscription immediately.
        */
        ClientMetricsInstance clientInstance = clientInstance(clientInstanceId, requestContext);

        try {
            // Validate the get request parameters for the client instance.
            validateGetRequest(request, clientInstance, now);
        } catch (ApiException exception) {
            return request.getErrorResponse(0, exception);
        }

        clientInstance.lastKnownError(Errors.NONE);
        return createGetSubscriptionResponse(clientInstanceId, clientInstance);
    }
@Test
    public void testGetTelemetryConcurrentRequestNewClientInstance() throws Exception {
        GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
            new GetTelemetrySubscriptionsRequestData().setClientInstanceId(Uuid.randomUuid()), true).build();

        CountDownLatch lock = new CountDownLatch(2);
        List<GetTelemetrySubscriptionsResponse> responses = Collections.synchronizedList(new ArrayList<>());

        Thread thread = new Thread(() -> {
            try {
                GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
                    request, ClientMetricsTestUtils.requestContext());

                responses.add(response);
            } catch (UnknownHostException e) {
                LOG.error("Error processing request", e);
            } finally {
                lock.countDown();
            }
        });

        Thread thread1 = new Thread(() -> {
            try {
                GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
                    request, ClientMetricsTestUtils.requestContext());

                responses.add(response);
            } catch (UnknownHostException e) {
                LOG.error("Error processing request", e);
            } finally {
                lock.countDown();
            }
        });

        thread.start();
        thread1.start();

        assertTrue(lock.await(2000, TimeUnit.MILLISECONDS));
        assertEquals(2, responses.size());

        int throttlingErrorCount = 0;
        for (GetTelemetrySubscriptionsResponse response : responses) {
            if (response.error() == Errors.THROTTLING_QUOTA_EXCEEDED) {
                throttlingErrorCount++;
            } else {
                // As subscription is updated hence 1 request shall fail with unknown subscription id.
                assertEquals(Errors.NONE, response.error());
            }
        }
        // 1 request should fail with throttling error.
        assertEquals(1, throttlingErrorCount);
        // Should register 1 throttle metric.
        assertEquals((double) 1, getMetric(ClientMetricsManager.ClientMetricsStats.THROTTLE + "-count").metricValue());
        assertTrue((double) getMetric(ClientMetricsManager.ClientMetricsStats.THROTTLE + "-rate").metricValue() > 0);
    }