TopicSerialization { public static Topic fromTopicResource(KafkaTopic kafkaTopic) { if (kafkaTopic == null) { return null; } Topic.Builder builder = new Topic.Builder() .withMapName(kafkaTopic.getMetadata().getName()) .withTopicName(getTopicName(kafkaTopic)) .withNumPartitions(getPartitions(kafkaTopic)) .withNumReplicas(getReplicas(kafkaTopic)) .withConfig(topicConfigFromTopicConfig(kafkaTopic)) .withMetadata(kafkaTopic.getMetadata()); 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 testErrorInDefaultTopicName() { String illegalAsATopicName = "012345678901234567890123456789012345678901234567890123456789" + "01234567890123456789012345678901234567890123456789012345678901234567890123456789" + "01234567890123456789012345678901234567890123456789012345678901234567890123456789" + "012345678901234567890123456789"; KafkaTopic kafkaTopic = new KafkaTopicBuilder().withMetadata(new ObjectMetaBuilder().withName(illegalAsATopicName) .build()).withNewSpec() .withReplicas(1) .withPartitions(1) .withConfig(emptyMap()) .endSpec() .build(); try { TopicSerialization.fromTopicResource(kafkaTopic); fail("Should throw"); } catch (InvalidTopicException e) { assertThat(e.getMessage(), is("KafkaTopics's spec.topicName property is absent and KafkaTopics's metadata.name is invalid as a topic name: " + "Topic name is illegal, it can't be longer than 249 characters, topic name: " + illegalAsATopicName)); } }
@Test public void testErrorInTopicName() { KafkaTopic kafkaTopic = new KafkaTopicBuilder().withMetadata(new ObjectMetaBuilder().withName("foo") .build()).withNewSpec() .withReplicas(1) .withPartitions(1) .withConfig(emptyMap()) .withTopicName("An invalid topic name!") .endSpec() .build(); try { TopicSerialization.fromTopicResource(kafkaTopic); fail("Should throw"); } catch (InvalidTopicException e) { assertThat(e.getMessage(), is("KafkaTopics's spec.topicName property is invalid as a topic name: Topic name \"An invalid topic name!\" is illegal, it contains a character other than ASCII alphanumerics, '.', '_' and '-'")); } }
@Test public void testErrorInPartitions() { KafkaTopic kafkaTopic = new KafkaTopicBuilder() .withMetadata(new ObjectMetaBuilder().withName("my-topic").build()) .withNewSpec() .withReplicas(1) .withPartitions(-1) .withConfig(emptyMap()) .endSpec() .build(); try { TopicSerialization.fromTopicResource(kafkaTopic); fail("Should throw"); } catch (InvalidTopicException e) { assertThat(e.getMessage(), is("KafkaTopic's spec.partitions should be strictly greater than 0")); } }
@Test public void testErrorInReplicas() { KafkaTopic kafkaTopic = new KafkaTopicBuilder() .withMetadata(new ObjectMetaBuilder().withName("my-topic").build()) .withNewSpec() .withReplicas(-1) .withPartitions(1) .withConfig(emptyMap()) .endSpec() .build(); try { TopicSerialization.fromTopicResource(kafkaTopic); fail("Should throw"); } catch (InvalidTopicException e) { assertThat(e.getMessage(), is("KafkaTopic's spec.replicas should be between 1 and 32767 inclusive")); } }
@Test public void testErrorInConfigInvalidValueWrongType() { KafkaTopic kafkaTopic = new KafkaTopicBuilder() .withMetadata(new ObjectMetaBuilder().withName("my-topic").build()) .withNewSpec() .withReplicas(1) .withPartitions(1) .withConfig(singletonMap("foo", new Object())) .endSpec() .build(); try { TopicSerialization.fromTopicResource(kafkaTopic); fail("Should throw"); } catch (InvalidTopicException e) { assertThat(e.getMessage(), is("KafkaTopic's spec.config has invalid entry: The key 'foo' of the topic config is invalid: The value corresponding to the key must have a string, number or boolean value but was of type java.lang.Object")); } }
@Test public void testErrorInConfigInvalidValueNull() { KafkaTopic kafkaTopic = new KafkaTopicBuilder() .withMetadata(new ObjectMetaBuilder().withName("my-topic").build()) .withNewSpec() .withReplicas(1) .withPartitions(1) .withConfig(singletonMap("foo", null)) .endSpec() .build(); try { TopicSerialization.fromTopicResource(kafkaTopic); fail("Should throw"); } catch (InvalidTopicException e) { assertThat(e.getMessage(), is("KafkaTopic's spec.config has invalid entry: The key 'foo' of the topic config is invalid: The value corresponding to the key must have a string, number or boolean value but the value was null")); } }
@Test public void testNoConfig() { KafkaTopicSpec spec = new KafkaTopicSpec(); spec.setPartitions(3); spec.setReplicas(3); KafkaTopic kafkaTopic = new KafkaTopic(); kafkaTopic.setMetadata(new ObjectMetaBuilder().withName("my-topic").build()); kafkaTopic.setSpec(spec); TopicSerialization.fromTopicResource(kafkaTopic); }