SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow underscore in kafka topic name

Browse Source 
- 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
This commit is contained in:
Ilayaperumal Gopinathan
2016-02-03 14:52:03 +05:30
committed by Mark Fisher
parent e9286c849a
commit 8f795074e9
2 changed files with 21 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -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<Partition> partitions = ensureTopicCreated(topicName, numPartitions, defaultReplicationFactor);
Collection<Partition> partitions = ensureTopicCreated(name, numPartitions, defaultReplicationFactor);
ProducerMetadata<byte[], byte[]> 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<byte[], byte[]> 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<Partition> allPartitions = ensureTopicCreated(topic, numPartitions, defaultReplicationFactor);
Collection<Partition> allPartitions = ensureTopicCreated(name, numPartitions, defaultReplicationFactor);
Decoder<byte[]> valueDecoder = new DefaultDecoder(null);
Decoder<byte[]> keyDecoder = new DefaultDecoder(null);

10
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -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<KafkaMessage> messages = new ArrayBlockingQueue<KafkaMessage>(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" };