TopicSerialization { public static NewTopic toNewTopic(Topic topic, Map<Integer, List<Integer>> assignment) { NewTopic newTopic; if (assignment != null) { if (topic.getNumPartitions() != assignment.size()) { throw new IllegalArgumentException( format("Topic %s has %d partitions supplied, but the number of partitions " + "configured in KafkaTopic %s is %d", topic.getTopicName(), assignment.size(), topic.getResourceName(), topic.getNumPartitions())); } for (int partition = 0; partition < assignment.size(); partition++) { final List<Integer> value = assignment.get(partition); if (topic.getNumReplicas() != value.size()) { throw new IllegalArgumentException( format("Partition %d of topic %s has %d assigned replicas, " + "but the number of replicas configured in KafkaTopic %s for the topic is %d", partition, topic.getTopicName(), value.size(), topic.getResourceName(), topic.getNumReplicas())); } } newTopic = new NewTopic(topic.getTopicName().toString(), assignment); } else { newTopic = new NewTopic(topic.getTopicName().toString(), topic.getNumPartitions(), topic.getNumReplicas()); } newTopic.configs(topic.getConfig()); return newTopic; }  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 testToNewTopic() { Topic topic = new Topic.Builder() .withTopicName("test-topic") .withConfigEntry("foo", "bar") .withNumPartitions(3) .withNumReplicas((short) 2) .withMapName("gee") .build(); NewTopic newTopic = TopicSerialization.toNewTopic(topic, null); assertThat(newTopic.name(), is("test-topic")); assertThat(newTopic.numPartitions(), is(3)); assertThat(newTopic.replicationFactor(), is((short) 2)); assertThat(newTopic.replicasAssignments(), is(nullValue())); assertThat(newTopic.configs(), is(singletonMap("foo", "bar"))); }