@Test public void getTopicConfigsTest() { this.propertySupplierMap.put(ClusterConfigKey.GET_TOPIC_DYNAMIC_METADATA_CACHE_ENABLE.getName(), true); clusterNameServiceStub.setSleep(1000 * 1); List<String> topics = Lists.newLinkedList(); for (int i = 2; i < 10; i++) { topics.add("test" + i); } long startTime = SystemClock.now(); Map<String, TopicConfig> topicConfigs = clusterNameServiceStub.getTopicConfigs(topics); Assert.assertEquals(true, SystemClock.now() - startTime >= 1000 * 2 && SystemClock.now() - startTime <= 1000 * 6); startTime = SystemClock.now(); clusterNameServiceStub.getTopicConfigs(topics); Assert.assertEquals(true, SystemClock.now() - startTime < 1000 * 5); this.propertySupplierMap.put(ClusterConfigKey.GET_TOPIC_DYNAMIC_METADATA_CACHE_ENABLE.getName(), false); this.propertySupplierMap.put(ClusterConfigKey.GET_TOPIC_DYNAMIC_METADATA_BATCH_PARALLEL_ENABLE.getName(), false); startTime = SystemClock.now(); clusterNameServiceStub.getTopicConfigs(topics); Assert.assertEquals(true, SystemClock.now() - startTime >= 1000 * 2 * 8); clusterNameServiceStub.setSleep(0); }
public Map<String, TopicConfig> getTopicConfigs(List<String> topics) { Map<String, TopicConfig> topicConfigMap = Maps.newHashMap(); for (String topic : topics) { TopicConfig topicConfig = nameService.getTopicConfig(TopicName.parse(topic)); if (topicConfig != null) { topicConfigMap.put(topic, topicConfig); } } if (MapUtils.isEmpty(topicConfigMap)) { return topicConfigMap; } if (topicConfigMap.size() > config.getTopicDynamicMetadataMaxBatchThreshold()) { return topicConfigMap; } TraceStat begin = tracer.begin("ClusterNameService.getTopicConfigs"); try { CountDownLatch latch = new CountDownLatch(topicConfigMap.size()); Map<String, TopicConfig> result = Maps.newConcurrentMap(); for (Map.Entry<String, TopicConfig> entry : topicConfigMap.entrySet()) { if (config.getTopicDynamicMetadataBatchParallelEnable()) { result.put(entry.getKey(), entry.getValue()); clusterNameServiceExecutorService.execute(() -> { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); }); } else { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); } } if (!latch.await(config.getTopicDynamicMetadataBatchTimeout(), TimeUnit.MILLISECONDS)) { logger.error("getTopicConfigs timeout, topics: {}", topics); } tracer.end(begin); return result; } catch (Exception e) { logger.error("getTopicConfigs exception, topics: {}", topics, e); tracer.error(begin); return topicConfigMap; } }
ClusterNameService extends Service { public Map<String, TopicConfig> getTopicConfigs(List<String> topics) { Map<String, TopicConfig> topicConfigMap = Maps.newHashMap(); for (String topic : topics) { TopicConfig topicConfig = nameService.getTopicConfig(TopicName.parse(topic)); if (topicConfig != null) { topicConfigMap.put(topic, topicConfig); } } if (MapUtils.isEmpty(topicConfigMap)) { return topicConfigMap; } if (topicConfigMap.size() > config.getTopicDynamicMetadataMaxBatchThreshold()) { return topicConfigMap; } TraceStat begin = tracer.begin("ClusterNameService.getTopicConfigs"); try { CountDownLatch latch = new CountDownLatch(topicConfigMap.size()); Map<String, TopicConfig> result = Maps.newConcurrentMap(); for (Map.Entry<String, TopicConfig> entry : topicConfigMap.entrySet()) { if (config.getTopicDynamicMetadataBatchParallelEnable()) { result.put(entry.getKey(), entry.getValue()); clusterNameServiceExecutorService.execute(() -> { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); }); } else { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); } } if (!latch.await(config.getTopicDynamicMetadataBatchTimeout(), TimeUnit.MILLISECONDS)) { logger.error("getTopicConfigs timeout, topics: {}", topics); } tracer.end(begin); return result; } catch (Exception e) { logger.error("getTopicConfigs exception, topics: {}", topics, e); tracer.error(begin); return topicConfigMap; } } }
ClusterNameService extends Service { public Map<String, TopicConfig> getTopicConfigs(List<String> topics) { Map<String, TopicConfig> topicConfigMap = Maps.newHashMap(); for (String topic : topics) { TopicConfig topicConfig = nameService.getTopicConfig(TopicName.parse(topic)); if (topicConfig != null) { topicConfigMap.put(topic, topicConfig); } } if (MapUtils.isEmpty(topicConfigMap)) { return topicConfigMap; } if (topicConfigMap.size() > config.getTopicDynamicMetadataMaxBatchThreshold()) { return topicConfigMap; } TraceStat begin = tracer.begin("ClusterNameService.getTopicConfigs"); try { CountDownLatch latch = new CountDownLatch(topicConfigMap.size()); Map<String, TopicConfig> result = Maps.newConcurrentMap(); for (Map.Entry<String, TopicConfig> entry : topicConfigMap.entrySet()) { if (config.getTopicDynamicMetadataBatchParallelEnable()) { result.put(entry.getKey(), entry.getValue()); clusterNameServiceExecutorService.execute(() -> { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); }); } else { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); } } if (!latch.await(config.getTopicDynamicMetadataBatchTimeout(), TimeUnit.MILLISECONDS)) { logger.error("getTopicConfigs timeout, topics: {}", topics); } tracer.end(begin); return result; } catch (Exception e) { logger.error("getTopicConfigs exception, topics: {}", topics, e); tracer.error(begin); return topicConfigMap; } } ClusterNameService(NameService nameService, BrokerEventBus eventBus, PropertySupplier propertySupplier); }
ClusterNameService extends Service { public Map<String, TopicConfig> getTopicConfigs(List<String> topics) { Map<String, TopicConfig> topicConfigMap = Maps.newHashMap(); for (String topic : topics) { TopicConfig topicConfig = nameService.getTopicConfig(TopicName.parse(topic)); if (topicConfig != null) { topicConfigMap.put(topic, topicConfig); } } if (MapUtils.isEmpty(topicConfigMap)) { return topicConfigMap; } if (topicConfigMap.size() > config.getTopicDynamicMetadataMaxBatchThreshold()) { return topicConfigMap; } TraceStat begin = tracer.begin("ClusterNameService.getTopicConfigs"); try { CountDownLatch latch = new CountDownLatch(topicConfigMap.size()); Map<String, TopicConfig> result = Maps.newConcurrentMap(); for (Map.Entry<String, TopicConfig> entry : topicConfigMap.entrySet()) { if (config.getTopicDynamicMetadataBatchParallelEnable()) { result.put(entry.getKey(), entry.getValue()); clusterNameServiceExecutorService.execute(() -> { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); }); } else { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); } } if (!latch.await(config.getTopicDynamicMetadataBatchTimeout(), TimeUnit.MILLISECONDS)) { logger.error("getTopicConfigs timeout, topics: {}", topics); } tracer.end(begin); return result; } catch (Exception e) { logger.error("getTopicConfigs exception, topics: {}", topics, e); tracer.error(begin); return topicConfigMap; } } ClusterNameService(NameService nameService, BrokerEventBus eventBus, PropertySupplier propertySupplier); void setBroker(Broker broker); ClusterNode getTopicGroupNode(TopicName topicName, int group); ClusterNode getTopicGroupNode(String topic, int group); Map<TopicName, TopicConfig> getTopicConfigByApp(String subscribeApp, Subscription.Type subscribe); Map<String, TopicConfig> getTopicConfigs(List<String> topics); TopicConfig getTopicConfig(TopicName topicName); TopicConfig doGetTopicConfig(TopicConfig topicConfig); NameService getNameService(); }
ClusterNameService extends Service { public Map<String, TopicConfig> getTopicConfigs(List<String> topics) { Map<String, TopicConfig> topicConfigMap = Maps.newHashMap(); for (String topic : topics) { TopicConfig topicConfig = nameService.getTopicConfig(TopicName.parse(topic)); if (topicConfig != null) { topicConfigMap.put(topic, topicConfig); } } if (MapUtils.isEmpty(topicConfigMap)) { return topicConfigMap; } if (topicConfigMap.size() > config.getTopicDynamicMetadataMaxBatchThreshold()) { return topicConfigMap; } TraceStat begin = tracer.begin("ClusterNameService.getTopicConfigs"); try { CountDownLatch latch = new CountDownLatch(topicConfigMap.size()); Map<String, TopicConfig> result = Maps.newConcurrentMap(); for (Map.Entry<String, TopicConfig> entry : topicConfigMap.entrySet()) { if (config.getTopicDynamicMetadataBatchParallelEnable()) { result.put(entry.getKey(), entry.getValue()); clusterNameServiceExecutorService.execute(() -> { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); }); } else { result.put(entry.getKey(), doGetTopicConfig(entry.getValue())); latch.countDown(); } } if (!latch.await(config.getTopicDynamicMetadataBatchTimeout(), TimeUnit.MILLISECONDS)) { logger.error("getTopicConfigs timeout, topics: {}", topics); } tracer.end(begin); return result; } catch (Exception e) { logger.error("getTopicConfigs exception, topics: {}", topics, e); tracer.error(begin); return topicConfigMap; } } ClusterNameService(NameService nameService, BrokerEventBus eventBus, PropertySupplier propertySupplier); void setBroker(Broker broker); ClusterNode getTopicGroupNode(TopicName topicName, int group); ClusterNode getTopicGroupNode(String topic, int group); Map<TopicName, TopicConfig> getTopicConfigByApp(String subscribeApp, Subscription.Type subscribe); Map<String, TopicConfig> getTopicConfigs(List<String> topics); TopicConfig getTopicConfig(TopicName topicName); TopicConfig doGetTopicConfig(TopicConfig topicConfig); NameService getNameService();  }