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

Use TopicExistsException in provisioner

Browse Source 
Instead of checking for the text TopicExistsException in the exception
message, use strong type check for TopicExistsException through instanceof on
the cause of the exception.

Adding test to verify.

Resolves #209
This commit is contained in:
Soby Chacko
2018-11-05 16:17:51 -05:00
committed by Artem Bilan
parent 84249727ae
commit 2d4e9a113b
2 changed files with 18 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
View File

@@ -39,6 +39,7 @@ import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.errors.TopicExistsException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
@@ -370,8 +371,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
}
catch (Exception ex) {
if (ex instanceof ExecutionException) {
String exceptionMessage = ex.getMessage();
if (exceptionMessage.contains("org.apache.kafka.common.errors.TopicExistsException")) {
if (ex.getCause() instanceof TopicExistsException) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("Attempt to create topic: " + topicName + ". Topic already exists.");
}

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

@@ -35,7 +35,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreateTopicsResult;
@@ -50,6 +49,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.errors.TopicExistsException;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
@@ -2649,6 +2649,21 @@ public class KafkaBinderTests extends
}
}
@Test (expected = TopicExistsException.class)
public void testSameTopicCannotBeProvisionedAgain() throws Throwable {
try (AdminClient admin = AdminClient.create(Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
embeddedKafka.getEmbeddedKafka().getBrokersAsString()))) {
admin.createTopics(Collections.singletonList(new NewTopic("fooUniqueTopic", 1, (short) 1))).all().get();
try {
admin.createTopics(Collections.singletonList(new NewTopic("fooUniqueTopic", 1, (short) 1))).all().get();
}
catch (Exception ex) {
assertThat(ex.getCause() instanceof TopicExistsException).isTrue();
throw ex.getCause();
}
}
}
private final class FailingInvocationCountingMessageHandler implements MessageHandler {
private int invocationCount;