From 8f795074e9983f471f92f179189ab31c61a4b3a0 Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Wed, 3 Feb 2016 14:52:03 +0530 Subject: [PATCH] Allow underscore in kafka topic name - Instead of escaping the topic name, perform validation that throws `RuntimeException` when the given kafka topic name doesn't meet the criteria set by Kafka. This resolves #217 --- .../kafka/KafkaMessageChannelBinder.java | 33 ++++++++----------- .../stream/binder/kafka/KafkaBinderTests.java | 10 ++++-- 2 files changed, 21 insertions(+), 22 deletions(-) diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 9cc69a56e..d4bc55797 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -37,11 +37,11 @@ import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.springframework.beans.factory.DisposableBean; -import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor; import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.BinderPropertyKeys; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor; import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter; import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; import org.springframework.cloud.stream.binder.MessageValues; @@ -376,28 +376,21 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { } /** - * Allowed chars are ASCII alphanumerics, '.', '_' and '-'. '_' is used as escaped char in the form '_xx' where xx - * is the hexadecimal value of the byte(s) needed to represent an illegal char in utf8. + * Allowed chars are ASCII alphanumerics, '.', '_' and '-'. */ - /*default*/ - public static String escapeTopicName(String original) { - StringBuilder result = new StringBuilder(original.length()); + public static void validateTopicName(String topicName) { try { - byte[] utf8 = original.getBytes("UTF-8"); + byte[] utf8 = topicName.getBytes("UTF-8"); for (byte b : utf8) { - if ((b >= 'a') && (b <= 'z') || (b >= 'A') && (b <= 'Z') || (b >= '0') && (b <= '9') || (b == '.') - || (b == '-')) { - result.append((char) b); - } - else { - result.append(String.format("_%02X", b)); + if (!((b >= 'a') && (b <= 'z') || (b >= 'A') && (b <= 'Z') || (b >= '0') && (b <= '9') || (b == '.') + || (b == '-') || (b == '_'))) { + throw new IllegalArgumentException("Topic name can only have ASCII alphanumerics, '.', '_' and '-'"); } } } catch (UnsupportedEncodingException e) { throw new AssertionError(e); // Can't happen } - return result.toString(); } public void setDefaultReplicationFactor(int defaultReplicationFactor) { @@ -461,14 +454,14 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { logger.info("Using kafka topic for outbound: " + name); } - final String topicName = escapeTopicName(name); + validateTopicName(name); int numPartitions = producerPropertiesAccessor.getNumberOfKafkaPartitionsForProducer(); - Collection partitions = ensureTopicCreated(topicName, numPartitions, defaultReplicationFactor); + Collection partitions = ensureTopicCreated(name, numPartitions, defaultReplicationFactor); ProducerMetadata producerMetadata = new ProducerMetadata<>( - topicName, byte[].class, byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER); + name, byte[].class, byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER); producerMetadata.setCompressionType(ProducerMetadata.CompressionType.valueOf( producerPropertiesAccessor.getCompressionCodec(this.defaultCompressionCodec))); producerMetadata.setBatchBytes(producerPropertiesAccessor.getBatchSize(this.defaultBatchSize)); @@ -486,7 +479,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { final ProducerConfiguration producerConfiguration = new ProducerConfiguration<>( producerMetadata, producerFB.getObject()); - MessageHandler handler = new SendingHandler(topicName, producerPropertiesAccessor, + MessageHandler handler = new SendingHandler(name, producerPropertiesAccessor, partitions.size(), producerConfiguration); EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel, handler); @@ -567,10 +560,10 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { int maxConcurrency = accessor.getConcurrency(defaultConcurrency); - String topic = escapeTopicName(name); + validateTopicName(name); int numPartitions = accessor.getNumberOfKafkaPartitionsForConsumer(); - Collection allPartitions = ensureTopicCreated(topic, numPartitions, defaultReplicationFactor); + Collection allPartitions = ensureTopicCreated(name, numPartitions, defaultReplicationFactor); Decoder valueDecoder = new DefaultDecoder(null); Decoder keyDecoder = new DefaultDecoder(null); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 0873b027e..f5648cb26 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -56,6 +56,7 @@ import kafka.api.OffsetRequest; * @author Eric Bottard * @author Marius Bogoevici * @author Mark Fisher + * @author Ilayaperumal Gopinathan */ public class KafkaBinderTests extends PartitionCapableBinderTests { @@ -95,13 +96,13 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { @Override public Spy spyOn(final String name) { - String topic = KafkaMessageChannelBinder.escapeTopicName(name); + KafkaMessageChannelBinder.validateTopicName(name); KafkaTestBinder binderWrapper = (KafkaTestBinder) getBinder(); // Rewind offset, as tests will have typically already sent the messages we're trying to consume KafkaMessageListenerContainer messageListenerContainer = binderWrapper.getCoreBinder().createMessageListenerContainer( - new Properties(), UUID.randomUUID().toString(), 1, topic, OffsetRequest.EarliestTime()); + new Properties(), UUID.randomUUID().toString(), 1, name, OffsetRequest.EarliestTime()); final BlockingQueue messages = new ArrayBlockingQueue(10); @@ -124,6 +125,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { } + @Test(expected = RuntimeException.class) + public void testValidateKafkaTopicName() { + KafkaMessageChannelBinder.validateTopicName("foo:bar"); + } + @Test public void testCompression() throws Exception { final String[] codecs = new String[] { null, "none", "gzip", "snappy" };