TopicSerialization { public static Topic fromTopicMetadata(TopicMetadata meta) { if (meta == null) { return null; } Topic.Builder builder = new Topic.Builder() .withTopicName(meta.getDescription().name()) .withNumPartitions(meta.getDescription().partitions().size()) .withNumReplicas((short) meta.getDescription().partitions().get(0).replicas().size()) .withMetadata(null); for (ConfigEntry entry: meta.getConfig().entries()) { if (entry.source() != ConfigEntry.ConfigSource.DEFAULT_CONFIG && entry.source() != ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG) { builder.withConfigEntry(entry.name(), entry.value()); } } return builder.build(); }  static Topic fromTopicResource(KafkaTopic kafkaTopic); static KafkaTopic toTopicResource(Topic topic, Labels labels); static NewTopic toNewTopic(Topic topic, Map<Integer, List<Integer>> assignment); static Map<ConfigResource, Config> toTopicConfig(Topic topic); static Topic fromTopicMetadata(TopicMetadata meta); static byte[] toJson(Topic topic); @SuppressWarnings("unchecked") static Topic fromJson(byte[] json); static final String JSON_KEY_TOPIC_NAME; static final String JSON_KEY_MAP_NAME; static final String JSON_KEY_PARTITIONS; static final String JSON_KEY_REPLICAS; static final String JSON_KEY_CONFIG; }
@Test public void testFromTopicMetadata() { List<ConfigEntry> entries = new ArrayList<>(); entries.add(new ConfigEntry("foo", "bar")); Config topicConfig = new Config(entries); TopicMetadata meta = Utils.getTopicMetadata("test-topic", topicConfig); Topic topic = TopicSerialization.fromTopicMetadata(meta); assertThat(topic.getTopicName(), is(new TopicName("test-topic"))); assertThat(topic.getResourceName(), is(nullValue())); assertThat(topic.getConfig(), is(singletonMap("foo", "bar"))); assertThat(topic.getNumPartitions(), is(2)); assertThat(topic.getNumReplicas(), is((short) 3)); }