@Test(expected = Exception.class) public void testValidateNonexistentTopic() throws Exception { assertFalse(topicEnsure.validateTopic(simpleTopicSpec("unknown-topic"), TIMEOUT_MS)); }
public boolean validateTopic(TopicSpec spec, int timeOut) throws Exception { DescribeTopicsResult topicDescribeResult = adminClient.describeTopics( Collections.singletonList(spec.name()), new DescribeTopicsOptions().timeoutMs(timeOut) ); TopicDescription topic = topicDescribeResult.all().get().get(spec.name()); ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, spec.name()); DescribeConfigsResult configResult = adminClient.describeConfigs( Collections.singletonList(configResource) ); Map<ConfigResource, Config> resultMap = configResult.all().get(); Config config = resultMap.get(configResource); Map<String, String> actualConfig = new HashMap<>(); for (Map.Entry<String, String> entry : spec.config().entrySet()) { ConfigEntry actualConfigEntry = config.get(entry.getKey()); if (actualConfigEntry != null) { actualConfig.put(entry.getKey(), actualConfigEntry.value()); } } TopicSpec actualSpec = new TopicSpec( topic.name(), topic.partitions().size(), topic.partitions().get(0).replicas().size(), actualConfig ); boolean isTopicValid = actualSpec.equals(spec); if (!isTopicValid) { System.err.printf( "Invalid topic [ %s ] ! Expected %s but got %s\n", spec.name(), spec, actualSpec ); } return isTopicValid; }
TopicEnsure { public boolean validateTopic(TopicSpec spec, int timeOut) throws Exception { DescribeTopicsResult topicDescribeResult = adminClient.describeTopics( Collections.singletonList(spec.name()), new DescribeTopicsOptions().timeoutMs(timeOut) ); TopicDescription topic = topicDescribeResult.all().get().get(spec.name()); ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, spec.name()); DescribeConfigsResult configResult = adminClient.describeConfigs( Collections.singletonList(configResource) ); Map<ConfigResource, Config> resultMap = configResult.all().get(); Config config = resultMap.get(configResource); Map<String, String> actualConfig = new HashMap<>(); for (Map.Entry<String, String> entry : spec.config().entrySet()) { ConfigEntry actualConfigEntry = config.get(entry.getKey()); if (actualConfigEntry != null) { actualConfig.put(entry.getKey(), actualConfigEntry.value()); } } TopicSpec actualSpec = new TopicSpec( topic.name(), topic.partitions().size(), topic.partitions().get(0).replicas().size(), actualConfig ); boolean isTopicValid = actualSpec.equals(spec); if (!isTopicValid) { System.err.printf( "Invalid topic [ %s ] ! Expected %s but got %s\n", spec.name(), spec, actualSpec ); } return isTopicValid; } }
TopicEnsure { public boolean validateTopic(TopicSpec spec, int timeOut) throws Exception { DescribeTopicsResult topicDescribeResult = adminClient.describeTopics( Collections.singletonList(spec.name()), new DescribeTopicsOptions().timeoutMs(timeOut) ); TopicDescription topic = topicDescribeResult.all().get().get(spec.name()); ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, spec.name()); DescribeConfigsResult configResult = adminClient.describeConfigs( Collections.singletonList(configResource) ); Map<ConfigResource, Config> resultMap = configResult.all().get(); Config config = resultMap.get(configResource); Map<String, String> actualConfig = new HashMap<>(); for (Map.Entry<String, String> entry : spec.config().entrySet()) { ConfigEntry actualConfigEntry = config.get(entry.getKey()); if (actualConfigEntry != null) { actualConfig.put(entry.getKey(), actualConfigEntry.value()); } } TopicSpec actualSpec = new TopicSpec( topic.name(), topic.partitions().size(), topic.partitions().get(0).replicas().size(), actualConfig ); boolean isTopicValid = actualSpec.equals(spec); if (!isTopicValid) { System.err.printf( "Invalid topic [ %s ] ! Expected %s but got %s\n", spec.name(), spec, actualSpec ); } return isTopicValid; } TopicEnsure(Properties props); }
TopicEnsure { public boolean validateTopic(TopicSpec spec, int timeOut) throws Exception { DescribeTopicsResult topicDescribeResult = adminClient.describeTopics( Collections.singletonList(spec.name()), new DescribeTopicsOptions().timeoutMs(timeOut) ); TopicDescription topic = topicDescribeResult.all().get().get(spec.name()); ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, spec.name()); DescribeConfigsResult configResult = adminClient.describeConfigs( Collections.singletonList(configResource) ); Map<ConfigResource, Config> resultMap = configResult.all().get(); Config config = resultMap.get(configResource); Map<String, String> actualConfig = new HashMap<>(); for (Map.Entry<String, String> entry : spec.config().entrySet()) { ConfigEntry actualConfigEntry = config.get(entry.getKey()); if (actualConfigEntry != null) { actualConfig.put(entry.getKey(), actualConfigEntry.value()); } } TopicSpec actualSpec = new TopicSpec( topic.name(), topic.partitions().size(), topic.partitions().get(0).replicas().size(), actualConfig ); boolean isTopicValid = actualSpec.equals(spec); if (!isTopicValid) { System.err.printf( "Invalid topic [ %s ] ! Expected %s but got %s\n", spec.name(), spec, actualSpec ); } return isTopicValid; } TopicEnsure(Properties props); boolean createTopic(TopicSpec spec, int timeOut); boolean validateTopic(TopicSpec spec, int timeOut); boolean topicExists(TopicSpec spec, Integer timeOut); }
TopicEnsure { public boolean validateTopic(TopicSpec spec, int timeOut) throws Exception { DescribeTopicsResult topicDescribeResult = adminClient.describeTopics( Collections.singletonList(spec.name()), new DescribeTopicsOptions().timeoutMs(timeOut) ); TopicDescription topic = topicDescribeResult.all().get().get(spec.name()); ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, spec.name()); DescribeConfigsResult configResult = adminClient.describeConfigs( Collections.singletonList(configResource) ); Map<ConfigResource, Config> resultMap = configResult.all().get(); Config config = resultMap.get(configResource); Map<String, String> actualConfig = new HashMap<>(); for (Map.Entry<String, String> entry : spec.config().entrySet()) { ConfigEntry actualConfigEntry = config.get(entry.getKey()); if (actualConfigEntry != null) { actualConfig.put(entry.getKey(), actualConfigEntry.value()); } } TopicSpec actualSpec = new TopicSpec( topic.name(), topic.partitions().size(), topic.partitions().get(0).replicas().size(), actualConfig ); boolean isTopicValid = actualSpec.equals(spec); if (!isTopicValid) { System.err.printf( "Invalid topic [ %s ] ! Expected %s but got %s\n", spec.name(), spec, actualSpec ); } return isTopicValid; } TopicEnsure(Properties props); boolean createTopic(TopicSpec spec, int timeOut); boolean validateTopic(TopicSpec spec, int timeOut); boolean topicExists(TopicSpec spec, Integer timeOut);  }