GH-1683: Refactoring

Simplify complexity.
Add nullable to ack.

Author: garyrussell

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

@@ -22,6 +22,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.lang.Nullable;
 
 /**
  * Listener for handling a batch of incoming Kafka messages, propagating an acknowledgment
@@ -50,6 +51,6 @@ default void onMessage(List<ConsumerRecord<K, V>> data) {
 	}
 
 	@Override
-	void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);
+	void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer);
 
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java

@@ -30,6 +30,7 @@
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.converter.BatchMessageConverter;
 import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
@@ -134,7 +135,9 @@ public void onMessage(ConsumerRecords<K, V> records, Acknowledgment acknowledgme
 	 * @param consumer the consumer.
 	 */
 	@Override
-	public void onMessage(List<ConsumerRecord<K, V>> records, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
+	public void onMessage(List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment acknowledgment,
+			Consumer<?, ?> consumer) {
+
 		Message<?> message;
 		if (!isConsumerRecordList()) {
 			if (isMessageList() || this.batchToRecordAdapter != null) {
@@ -162,7 +165,7 @@ public void onMessage(List<ConsumerRecord<K, V>> records, Acknowledgment acknowl
 		invoke(records, acknowledgment, consumer, message);
 	}
 
-	protected void invoke(Object records, Acknowledgment acknowledgment, Consumer<?, ?> consumer,
+	protected void invoke(Object records, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,
 			final Message<?> messageArg) {
 
 		Message<?> message = messageArg;
@@ -196,7 +199,9 @@ protected void invoke(Object records, Acknowledgment acknowledgment, Consumer<?,
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	protected Message<?> toMessagingMessage(List records, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
+	protected Message<?> toMessagingMessage(List records, @Nullable Acknowledgment acknowledgment,
+			Consumer<?, ?> consumer) {
+
 		return getBatchMessageConverter().toMessage(records, acknowledgment, consumer, getType());
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

@@ -300,7 +300,7 @@ public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekC
 		}
 	}
 
-	protected Message<?> toMessagingMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment,
+	protected Message<?> toMessagingMessage(ConsumerRecord<K, V> record, @Nullable Acknowledgment acknowledgment,
 			Consumer<?, ?> consumer) {
 		return getMessageConverter().toMessage(record, acknowledgment, consumer, getType());
 	}
@@ -314,7 +314,7 @@ protected Message<?> toMessagingMessage(ConsumerRecord<K, V> record, Acknowledgm
 	 * @param consumer the consumer.
 	 * @return the result of invocation.
 	 */
-	protected final Object invokeHandler(Object data, Acknowledgment acknowledgment, Message<?> message,
+	protected final Object invokeHandler(Object data, @Nullable Acknowledgment acknowledgment, Message<?> message,
 			Consumer<?, ?> consumer) {
 
 		try {
@@ -332,25 +332,10 @@ protected final Object invokeHandler(Object data, Acknowledgment acknowledgment,
 			}
 		}
 		catch (org.springframework.messaging.converter.MessageConversionException ex) {
-			if (this.hasAckParameter && acknowledgment == null) {
-				throw new ListenerExecutionFailedException("invokeHandler Failed",
-						new IllegalStateException("No Acknowledgment available as an argument, "
-								+ "the listener container must have a MANUAL AckMode to populate the Acknowledgment.",
-								ex));
-			}
-			throw new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " +
-					"be invoked with the incoming message", message.getPayload()),
-					new MessageConversionException("Cannot handle message", ex));
+			throw checkAckArg(acknowledgment, message, new MessageConversionException("Cannot handle message", ex));
 		}
 		catch (MethodArgumentNotValidException ex) {
-			if (this.hasAckParameter && acknowledgment == null) {
-				throw new ListenerExecutionFailedException("invokeHandler Failed",
-						new IllegalStateException("No Acknowledgment available as an argument, "
-								+ "the listener container must have a MANUAL AckMode to populate the Acknowledgment.",
-								ex));
-			}
-			throw new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " +
-					"be invoked with the incoming message", message.getPayload()), ex);
+			throw checkAckArg(acknowledgment, message, ex);
 		}
 		catch (MessagingException ex) {
 			throw new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " +
@@ -362,6 +347,17 @@ protected final Object invokeHandler(Object data, Acknowledgment acknowledgment,
 		}
 	}
 
+	private RuntimeException checkAckArg(@Nullable Acknowledgment acknowledgment, Message<?> message, Exception ex) {
+		if (this.hasAckParameter && acknowledgment == null) {
+			return new ListenerExecutionFailedException("invokeHandler Failed",
+					new IllegalStateException("No Acknowledgment available as an argument, "
+							+ "the listener container must have a MANUAL AckMode to populate the Acknowledgment.",
+							ex));
+		}
+		return new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " +
+				"be invoked with the incoming message", message.getPayload()), ex);
+	}
+
 	/**
 	 * Handle the given result object returned from the listener method, sending a
 	 * response message to the SendTo topic.

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java

@@ -25,6 +25,7 @@
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.listener.ListenerExecutionFailedException;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
 
@@ -71,7 +72,9 @@ public RecordMessagingMessageListenerAdapter(Object bean, Method method, KafkaLi
 	 * @param consumer the consumer.
 	 */
 	@Override
-	public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
+	public void onMessage(ConsumerRecord<K, V> record, @Nullable Acknowledgment acknowledgment,
+			Consumer<?, ?> consumer) {
+
 		Message<?> message;
 		if (isConversionNeeded()) {
 			message = toMessagingMessage(record, acknowledgment, consumer);