@NonNullApi for core package

Also use a single `Duration` object for polling.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/ConsumerFactory.java

@@ -21,6 +21,8 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import org.springframework.lang.Nullable;
+
 /**
  * The strategy to produce a {@link Consumer} instance(s).
  *
@@ -47,7 +49,7 @@ default Consumer<K, V> createConsumer() {
 	 * @return the consumer.
 	 * @since 1.3
 	 */
-	default Consumer<K, V> createConsumer(String clientIdSuffix) {
+	default Consumer<K, V> createConsumer(@Nullable String clientIdSuffix) {
 		return createConsumer(null, clientIdSuffix);
 	}
 
@@ -60,7 +62,7 @@ default Consumer<K, V> createConsumer(String clientIdSuffix) {
 	 * @return the consumer.
 	 * @since 1.3
 	 */
-	default Consumer<K, V> createConsumer(String groupId, String clientIdSuffix) {
+	default Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdSuffix) {
 		return createConsumer(groupId, null, clientIdSuffix);
 	}
 
@@ -76,7 +78,8 @@ default Consumer<K, V> createConsumer(String groupId, String clientIdSuffix) {
 	 * @return the consumer.
 	 * @since 2.1.1
 	 */
-	Consumer<K, V> createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix);
+	Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
+			@Nullable String clientIdSuffix);
 
 	/**
 	 * Return true if consumers created by this factory use auto commit.

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java

@@ -71,11 +71,11 @@ public DefaultKafkaConsumerFactory(Map<String, Object> configs,
 		this.valueDeserializer = valueDeserializer;
 	}
 
-	public void setKeyDeserializer(Deserializer<K> keyDeserializer) {
+	public void setKeyDeserializer(@Nullable Deserializer<K> keyDeserializer) {
 		this.keyDeserializer = keyDeserializer;
 	}
 
-	public void setValueDeserializer(Deserializer<V> valueDeserializer) {
+	public void setValueDeserializer(@Nullable Deserializer<V> valueDeserializer) {
 		this.valueDeserializer = valueDeserializer;
 	}
 
@@ -95,12 +95,14 @@ public Deserializer<V> getValueDeserializer() {
 	}
 
 	@Override
-	public Consumer<K, V> createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) {
+	public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
+			@Nullable String clientIdSuffix) {
+
 		return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix);
 	}
 
-	protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientIdPrefix,
-			String clientIdSuffix) {
+	protected KafkaConsumer<K, V> createKafkaConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
+			@Nullable String clientIdSuffix) {
 
 		boolean overrideClientIdPrefix = StringUtils.hasText(clientIdPrefix);
 		if (clientIdSuffix == null) {

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -116,11 +116,11 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 		this.valueSerializer = valueSerializer;
 	}
 
-	public void setKeySerializer(Serializer<K> keySerializer) {
+	public void setKeySerializer(@Nullable Serializer<K> keySerializer) {
 		this.keySerializer = keySerializer;
 	}
 
-	public void setValueSerializer(Serializer<V> valueSerializer) {
+	public void setValueSerializer(@Nullable Serializer<V> valueSerializer) {
 		this.valueSerializer = valueSerializer;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -38,6 +38,7 @@
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
@@ -126,7 +127,7 @@ public void setDefaultTopic(String defaultTopic) {
 	 * which logs errors only.
 	 * @param producerListener the listener; may be {@code null}.
 	 */
-	public void setProducerListener(ProducerListener<K, V> producerListener) {
+	public void setProducerListener(@Nullable ProducerListener<K, V> producerListener) {
 		this.producerListener = producerListener;
 	}
 
@@ -143,6 +144,7 @@ public MessageConverter getMessageConverter() {
 	 * @param messageConverter the message converter.
 	 */
 	public void setMessageConverter(RecordMessageConverter messageConverter) {
+		Assert.notNull(messageConverter, "'messageConverter' cannot be null");
 		this.messageConverter = messageConverter;
 	}
 
@@ -157,50 +159,51 @@ public boolean isTransactional() {
 	}
 
 	@Override
-	public ListenableFuture<SendResult<K, V>> sendDefault(V data) {
+	public ListenableFuture<SendResult<K, V>> sendDefault(@Nullable V data) {
 		return send(this.defaultTopic, data);
 	}
 
 	@Override
-	public ListenableFuture<SendResult<K, V>> sendDefault(K key, V data) {
+	public ListenableFuture<SendResult<K, V>> sendDefault(K key, @Nullable V data) {
 		return send(this.defaultTopic, key, data);
 	}
 
 	@Override
-	public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data) {
+	public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, @Nullable V data) {
 		return send(this.defaultTopic, partition, key, data);
 	}
 
 	@Override
-	public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data) {
+	public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, @Nullable V data) {
 		return send(this.defaultTopic, partition, timestamp, key, data);
 	}
 
 	@Override
-	public ListenableFuture<SendResult<K, V>> send(String topic, V data) {
+	public ListenableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {
 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);
 		return doSend(producerRecord);
 	}
 
 	@Override
-	public ListenableFuture<SendResult<K, V>> send(String topic, K key, V data) {
+	public ListenableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {
 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, data);
 		return doSend(producerRecord);
 	}
 
 	@Override
-	public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data) {
+	public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, @Nullable V data) {
 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);
 		return doSend(producerRecord);
 	}
 
 	@Override
-	public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data) {
+	public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key,
+			@Nullable V data) {
+
 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, timestamp, key, data);
 		return doSend(producerRecord);
 	}
 
-
 	@Override
 	public ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record) {
 		return doSend(record);
@@ -244,6 +247,7 @@ public List<PartitionInfo> partitionsFor(String topic) {
 
 	@Override
 	public <T> T execute(ProducerCallback<K, V, T> callback) {
+		Assert.notNull(callback, "'callback' cannot be null");
 		Producer<K, V> producer = getTheProducer();
 		try {
 			return callback.doInKafka(producer);
@@ -255,6 +259,7 @@ public <T> T execute(ProducerCallback<K, V, T> callback) {
 
 	@Override
 	public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
+		Assert.notNull(callback, "'callback' cannot be null");
 		Assert.state(this.transactional, "Producer factory does not support transactions");
 		Producer<K, V> producer = this.producers.get();
 		Assert.state(producer == null, "Nested calls to 'executeInTransaction' are not allowed");

spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactoryUtils.java

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.clients.producer.Producer;
 
+import org.springframework.lang.Nullable;
 import org.springframework.transaction.support.ResourceHolderSynchronization;
 import org.springframework.transaction.support.TransactionSynchronization;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -72,7 +73,7 @@ public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(
 		return resourceHolder;
 	}
 
-	public static <K, V> void releaseResources(KafkaResourceHolder<K, V> resourceHolder) {
+	public static <K, V> void releaseResources(@Nullable KafkaResourceHolder<K, V> resourceHolder) {
 		if (resourceHolder != null) {
 			resourceHolder.getProducer().close();
 		}

spring-kafka/src/main/java/org/springframework/kafka/core/package-info.java

@@ -1,4 +1,5 @@
 /**
  * Package for kafka core components
  */
+@org.springframework.lang.NonNullApi
 package org.springframework.kafka.core;

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

@@ -406,6 +406,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 		private final LogIfLevelEnabled commitLogger = new LogIfLevelEnabled(this.logger,
 				this.containerProperties.getCommitLogLevel());
 
+		private final Duration pollTimeout = Duration.ofMillis(this.containerProperties.getPollTimeout());
+
 		private volatile Map<TopicPartition, OffsetMetadata> definedPartitions;
 
 		private volatile Collection<TopicPartition> assignedPartitions;
@@ -705,8 +707,7 @@ public void run() {
 						}
 						publishConsumerPausedEvent(this.consumer.assignment());
 					}
-					ConsumerRecords<K, V> records = this.consumer
-							.poll(Duration.ofMillis(this.containerProperties.getPollTimeout()));
+					ConsumerRecords<K, V> records = this.consumer.poll(this.pollTimeout);
 					this.lastPoll = System.currentTimeMillis();
 					if (this.consumerPaused && !isPaused()) {
 						if (this.logger.isDebugEnabled()) {