Addressing PR review

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

Author: sobychacko

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

@@ -176,8 +176,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;
-		messageListener = createMessageListenerInstance(messageConverter);
+		final MessagingMessageListenerAdapter<K, V> messageListener = createMessageListenerInstance(messageConverter);
 		messageListener.setHandlerMethod(configureListenerAdapter(messageListener));
 		JavaUtils.INSTANCE
 			.acceptIfNotNull(getReplyTopic(), replyTopic -> {
@@ -203,6 +202,11 @@ protected HandlerAdapter configureListenerAdapter(MessagingMessageListenerAdapte
 		return new HandlerAdapter(invocableHandlerMethod);
 	}
 
+	/**
+	 * Create an empty {@link MessagingMessageListenerAdapter} instance.
+	 * @param messageConverter the converter (may be null).
+	 * @return the {@link MessagingMessageListenerAdapter} instance.
+	 */
 	protected MessagingMessageListenerAdapter<K, V> createMessageListenerInstance(
 			@Nullable MessageConverter messageConverter) {
 

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

@@ -186,12 +186,7 @@ else if (this.autoStartup != null) {
 
 	@Override
 	public ShareKafkaMessageListenerContainer<K, V> createContainer(TopicPartitionOffset... topicPartitions) {
-		return createContainerInstance(new KafkaListenerEndpointAdapter() {
-			@Override
-			public TopicPartitionOffset[] getTopicPartitionsToAssign() {
-				return Arrays.copyOf(topicPartitions, topicPartitions.length);
-			}
-		});
+		throw new UnsupportedOperationException("ShareConsumer does not support explicit partition assignment");
 	}
 
 	@Override
@@ -206,12 +201,7 @@ public Collection<String> getTopics() {
 
 	@Override
 	public ShareKafkaMessageListenerContainer<K, V> createContainer(Pattern topicPattern) {
-		return createContainerInstance(new KafkaListenerEndpointAdapter() {
-			@Override
-			public Pattern getTopicPattern() {
-				return topicPattern;
-			}
-		});
+		throw new UnsupportedOperationException("ShareConsumer does not support topic patterns");
 	}
 
 	/**
@@ -220,21 +210,9 @@ public Pattern getTopicPattern() {
 	 * @return the container instance
 	 */
 	protected ShareKafkaMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
-		TopicPartitionOffset[] topicPartitions = endpoint.getTopicPartitionsToAssign();
-		if (topicPartitions != null && topicPartitions.length > 0) {
-			return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(), new ContainerProperties(topicPartitions));
-		}
-		else {
-			Collection<String> topics = endpoint.getTopics();
-			Assert.state(topics != null, "'topics' must not be null");
-			if (!topics.isEmpty()) {
-				return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(),
-						new ContainerProperties(topics.toArray(new String[0])));
-			}
-			else {
-				return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(),
-						new ContainerProperties(endpoint.getTopicPattern()));
-			}
-		}
+		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])));
 	}
 }

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

@@ -172,7 +172,6 @@ private class ShareListenerConsumer implements Runnable {
 
 			this.genericListener = listener;
 			this.clientId = ShareKafkaMessageListenerContainer.this.getClientId();
-			// Subscribe to topics, just like in the test
 			ContainerProperties containerProperties = getContainerProperties();
 			this.consumer.subscribe(Arrays.asList(containerProperties.getTopics()));
 		}