From b5a0013e1e83516bb50ec556aa2837229ec62719 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@pivotal.io>
Date: Thu, 8 Mar 2018 09:11:06 -0500
Subject: [PATCH] GH-330 Polishing

Resolves #330
Resolves #334
---
 .../provisioning/KafkaTopicProvisioner.java   | 33 +++++++------------
 .../stream/binder/kafka/AdminConfigTests.java | 20 ++---------
 2 files changed, 14 insertions(+), 39 deletions(-)

diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
index a4bd6424f..f1aaaec03 100644
--- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
+++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
@@ -69,6 +69,7 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  * @author Ilayaperumal Gopinathan
  * @author Simon Flandergan
+ * @author Oleg Zhurakousky
  */
 public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsumerProperties<KafkaConsumerProperties>,
 		ExtendedProducerProperties<KafkaProducerProperties>>, InitializingBean {
@@ -128,23 +129,22 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		KafkaTopicUtils.validateTopicName(name);
 		try (AdminClient adminClient = AdminClient.create(this.adminClientProperties)) {
 			createTopic(adminClient, name, properties.getPartitionCount(), false, properties.getExtension().getAdmin());
+			int partitions = 0;
 			if (this.configurationProperties.isAutoCreateTopics()) {
 				DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(name));
 				KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult.all();
 
+				Map<String, TopicDescription> topicDescriptions = null;
 				try {
-					Map<String, TopicDescription> topicDescriptions = all.get(this.operationTimeout, TimeUnit.SECONDS);
-					TopicDescription topicDescription = topicDescriptions.get(name);
-					int partitions = topicDescription.partitions().size();
-					return new KafkaProducerDestination(name, partitions);
+					topicDescriptions = all.get(this.operationTimeout, TimeUnit.SECONDS);
 				}
 				catch (Exception e) {
 					throw new ProvisioningException("Problems encountered with partitions finding", e);
 				}
+				TopicDescription topicDescription = topicDescriptions.get(name);
+				partitions = topicDescription.partitions().size();
 			}
-			else {
-				return new KafkaProducerDestination(name);
-			}
+			return new KafkaProducerDestination(name, partitions);
 		}
 	}
 
@@ -160,6 +160,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			throw new IllegalArgumentException("Instance count cannot be zero");
 		}
 		int partitionCount = properties.getInstanceCount() * properties.getConcurrency();
+		ConsumerDestination consumerDestination = new KafkaConsumerDestination(name);
 		try (AdminClient adminClient = createAdminClient()) {
 			createTopic(adminClient, name, partitionCount, properties.getExtension().isAutoRebalanceEnabled(),
 					properties.getExtension().getAdmin());
@@ -170,19 +171,17 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 					Map<String, TopicDescription> topicDescriptions = all.get(operationTimeout, TimeUnit.SECONDS);
 					TopicDescription topicDescription = topicDescriptions.get(name);
 					int partitions = topicDescription.partitions().size();
-					ConsumerDestination dlqTopic = createDlqIfNeedBe(adminClient, name, group, properties, anonymous,
-							partitions);
-					if (dlqTopic != null) {
-						return dlqTopic;
+					consumerDestination = createDlqIfNeedBe(adminClient, name, group, properties, anonymous, partitions);
+					if (consumerDestination == null) {
+						consumerDestination = new KafkaConsumerDestination(name, partitions);
 					}
-					return new KafkaConsumerDestination(name, partitions);
 				}
 				catch (Exception e) {
 					throw new ProvisioningException("provisioning exception", e);
 				}
 			}
 		}
-		return new KafkaConsumerDestination(name);
+		return consumerDestination;
 	}
 
 	AdminClient createAdminClient() {
@@ -399,10 +398,6 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 
 		private final int partitions;
 
-		KafkaProducerDestination(String destinationName) {
-			this(destinationName, 0);
-		}
-
 		KafkaProducerDestination(String destinationName, Integer partitions) {
 			this.producerDestinationName = destinationName;
 			this.partitions = partitions;
@@ -454,10 +449,6 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			return this.consumerDestinationName;
 		}
 
-		public String getDlqName() {
-			return dlqName;
-		}
-
 		@Override
 		public String toString() {
 			return "KafkaConsumerDestination{" +
diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java
index 39ff7433e..a6b8c61e3 100644
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java
@@ -22,15 +22,12 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaAdminProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
 import org.springframework.cloud.stream.config.BinderFactoryConfiguration;
 import org.springframework.cloud.stream.config.BindingServiceConfiguration;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit4.SpringRunner;
@@ -43,20 +40,19 @@ import static org.assertj.core.api.Assertions.assertThat;
  *
  */
 @RunWith(SpringRunner.class)
-@SpringBootTest(classes = { AdminConfigTests.KafkaBinderConfig.class,
-		KafkaBinderConfiguration.class,
+@SpringBootTest(classes = {KafkaBinderConfiguration.class,
 		BinderFactoryConfiguration.class,
 		BindingServiceConfiguration.class })
 @TestPropertySource(properties = {
 		"spring.cloud.stream.kafka.bindings.input.consumer.admin.replication-factor=2",
 		"spring.cloud.stream.kafka.bindings.input.consumer.admin.replicas-assignments.0=0,1",
 		"spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0" })
+@EnableIntegration
 public class AdminConfigTests {
 
 	@Autowired
 	private KafkaMessageChannelBinder binder;
 
-	@SuppressWarnings("unchecked")
 	@Test
 	public void testProps() {
 		KafkaConsumerProperties consumerProps = this.binder.getExtendedConsumerProperties("input");
@@ -65,16 +61,4 @@ public class AdminConfigTests {
 		assertThat(admin.getReplicasAssignments().get(0)).isEqualTo(Arrays.asList(0, 1));
 		assertThat(admin.getConfiguration().get("message.format.version")).isEqualTo("0.9.0.0");
 	}
-
-	@Configuration
-	@EnableIntegration
-	public static class KafkaBinderConfig {
-
-		@Bean
-		public KafkaProperties kafkaProperties() {
-			return new KafkaProperties();
-		}
-
-	}
-
 }