Addressing PR review

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java

@@ -175,8 +175,7 @@ protected MessagingMessageListenerAdapter<K, V> createMessageListener(MessageLis
 
 		Assert.state(this.messageHandlerMethodFactory != null,
 				"Could not create message listener - MessageHandlerMethodFactory not set");
-
-		final MessagingMessageListenerAdapter<K, V> messageListener = createMessageListenerInstance(messageConverter);
+		MessagingMessageListenerAdapter<K, V> messageListener = createMessageListenerInstance(messageConverter);
 		messageListener.setHandlerMethod(configureListenerAdapter(messageListener));
 		JavaUtils.INSTANCE
 			.acceptIfNotNull(getReplyTopic(), replyTopic -> {

spring-kafka/src/main/java/org/springframework/kafka/config/ShareKafkaListenerContainerFactory.java

@@ -83,14 +83,6 @@ public void setApplicationContext(ApplicationContext applicationContext) {
 		this.applicationContext = applicationContext;
 	}
 
-	/**
-	 * Set the share consumer factory to use for creating containers.
-	 * @param shareConsumerFactory the share consumer factory
-	 */
-	public void setShareConsumerFactory(ShareConsumerFactory<? super K, ? super V> shareConsumerFactory) {
-		this.shareConsumerFactory = shareConsumerFactory;
-	}
-
 	/**
 	 * Get the share consumer factory.
 	 * @return the share consumer factory
@@ -99,6 +91,14 @@ public void setShareConsumerFactory(ShareConsumerFactory<? super K, ? super V> s
 		return this.shareConsumerFactory;
 	}
 
+	/**
+	 * Set the share consumer factory to use for creating containers.
+	 * @param shareConsumerFactory the share consumer factory
+	 */
+	public void setShareConsumerFactory(ShareConsumerFactory<? super K, ? super V> shareConsumerFactory) {
+		this.shareConsumerFactory = shareConsumerFactory;
+	}
+
 	/**
 	 * Set whether containers created by this factory should auto-start.
 	 * @param autoStartup true to auto-start
@@ -136,7 +136,8 @@ public ShareKafkaMessageListenerContainer<K, V> createListenerContainer(KafkaLis
 		if (endpoint instanceof AbstractKafkaListenerEndpoint) {
 			configureEndpoint((AbstractKafkaListenerEndpoint<K, V>) endpoint);
 		}
-		endpoint.setupListenerContainer(instance, null); // No message converter for MVP
+		// TODO: No message converter for queue at the moment
+		endpoint.setupListenerContainer(instance, null);
 		initializeContainer(instance, endpoint);
 		return instance;
 	}
@@ -192,6 +193,7 @@ public ShareKafkaMessageListenerContainer<K, V> createContainer(TopicPartitionOf
 	@Override
 	public ShareKafkaMessageListenerContainer<K, V> createContainer(String... topics) {
 		return createContainerInstance(new KafkaListenerEndpointAdapter() {
+
 			@Override
 			public Collection<String> getTopics() {
 				return Arrays.asList(topics);
@@ -213,6 +215,7 @@ protected ShareKafkaMessageListenerContainer<K, V> createContainerInstance(Kafka
 		Collection<String> topics = endpoint.getTopics();
 		Assert.state(topics != null, "'topics' must not be null");
 		return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(),
-					new ContainerProperties(topics.toArray(new String[0])));
+				new ContainerProperties(topics.toArray(new String[0])));
 	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainer.java

@@ -78,14 +78,6 @@ public ShareKafkaMessageListenerContainer(ShareConsumerFactory<? super K, ? supe
 		Assert.notNull(shareConsumerFactory, "A ShareConsumerFactory must be provided");
 	}
 
-	/**
-	 * Set the {@code client.id} to use for the consumer.
-	 * @param clientId the client id to set
-	 */
-	public void setClientId(String clientId) {
-		this.clientId = clientId;
-	}
-
 	/**
 	 * Get the {@code client.id} for the consumer.
 	 * @return the client id, or null if not set
@@ -95,6 +87,14 @@ public String getClientId() {
 		return this.clientId;
 	}
 
+	/**
+	 * Set the {@code client.id} to use for the consumer.
+	 * @param clientId the client id to set
+	 */
+	public void setClientId(String clientId) {
+		this.clientId = clientId;
+	}
+
 	@Override
 	public boolean isInExpectedState() {
 		return isRunning();
@@ -167,8 +167,8 @@ private class ShareListenerConsumer implements Runnable {
 
 		ShareListenerConsumer(GenericMessageListener<?> listener) {
 			this.consumer = ShareKafkaMessageListenerContainer.this.shareConsumerFactory.createShareConsumer(
-				ShareKafkaMessageListenerContainer.this.getGroupId(),
-				ShareKafkaMessageListenerContainer.this.getClientId());
+					ShareKafkaMessageListenerContainer.this.getGroupId(),
+					ShareKafkaMessageListenerContainer.this.getClientId());
 
 			this.genericListener = listener;
 			this.clientId = ShareKafkaMessageListenerContainer.this.getClientId();
@@ -196,7 +196,7 @@ public void run() {
 							}
 							else {
 								GenericMessageListener<ConsumerRecord<K, V>> listener =
-									(GenericMessageListener<ConsumerRecord<K, V>>) this.genericListener;
+										(GenericMessageListener<ConsumerRecord<K, V>>) this.genericListener;
 								listener.onMessage(record);
 							}
 							// Temporarily auto-acknowledge and commit.
@@ -239,9 +239,11 @@ private void wrapUp() {
 		@Override
 		public String toString() {
 			return "ShareKafkaMessageListenerContainer.ShareListenerConsumer ["
-				+ "consumerGroupId=" + this.consumerGroupId
-				+ ", clientId=" + this.clientId
-				+ "]";
+					+ "consumerGroupId=" + this.consumerGroupId
+					+ ", clientId=" + this.clientId
+					+ "]";
 		}
+
 	}
+
 }