public void updateSubscription(String subscriptionName, Properties properties) {
        // Validate the subscription properties.
        ClientMetricsConfigs.validate(subscriptionName, properties);
        // IncrementalAlterConfigs API will send empty configs when all the configs are deleted
        // for respective subscription. In that case, we need to remove the subscription from the map.
        if (properties.isEmpty()) {
            // Remove the subscription from the map if it exists, else ignore the config update.
            if (subscriptionMap.containsKey(subscriptionName)) {
                log.info("Removing subscription [{}] from the subscription map", subscriptionName);
                subscriptionMap.remove(subscriptionName);
                subscriptionUpdateVersion.incrementAndGet();
            }
            return;
        }

        updateClientSubscription(subscriptionName, new ClientMetricsConfigs(properties));
        /*
         Increment subscription update version to indicate that there is a change in the subscription. This will
         be used to determine if the next telemetry request needs to re-evaluate the subscription id
         as per the changed subscriptions.
        */
        subscriptionUpdateVersion.incrementAndGet();
    }
@Test
    public void testUpdateSubscription() throws Exception {
        assertTrue(clientMetricsManager.subscriptions().isEmpty());

        assertEquals(0, clientMetricsManager.subscriptionUpdateVersion());
        clientMetricsManager.updateSubscription("sub-1", ClientMetricsTestUtils.defaultProperties());

        assertEquals(1, clientMetricsManager.subscriptions().size());
        assertNotNull(clientMetricsManager.subscriptionInfo("sub-1"));

        ClientMetricsManager.SubscriptionInfo subscriptionInfo = clientMetricsManager.subscriptionInfo("sub-1");
        Set<String> metrics = subscriptionInfo.metrics();

        // Validate metrics.
        assertEquals(ClientMetricsTestUtils.DEFAULT_METRICS.split(",").length, metrics.size());
        Arrays.stream(ClientMetricsTestUtils.DEFAULT_METRICS.split(",")).forEach(metric ->
            assertTrue(metrics.contains(metric)));
        // Validate push interval.
        assertEquals(ClientMetricsTestUtils.defaultProperties().getProperty(ClientMetricsConfigs.PUSH_INTERVAL_MS),
            String.valueOf(subscriptionInfo.intervalMs()));

        // Validate match patterns.
        assertEquals(ClientMetricsTestUtils.DEFAULT_CLIENT_MATCH_PATTERNS.size(),
            subscriptionInfo.matchPattern().size());
        ClientMetricsTestUtils.DEFAULT_CLIENT_MATCH_PATTERNS.forEach(pattern -> {
            String[] split = pattern.split("=");
            assertTrue(subscriptionInfo.matchPattern().containsKey(split[0]));
            assertEquals(split[1], subscriptionInfo.matchPattern().get(split[0]).pattern());
        });
        assertEquals(1, clientMetricsManager.subscriptionUpdateVersion());
        // Validate metrics should have instance count metric, 2 unknown subscription count metrics
        // and kafka metrics count registered i.e. 4 metrics.
        assertEquals(4, kafkaMetrics.metrics().size());
        // Metrics should not have any instance while updating the subscriptions.
        assertEquals((double) 0, getMetric(ClientMetricsManager.ClientMetricsStats.INSTANCE_COUNT).metricValue());
    }