@NonNullApi for core package

garyrussell · artembilan · commit 82b816d0aeb2 · 2018-08-02T11:15:42.000-04:00
Also use a single `Duration` object for polling.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/ConsumerFactory.java b/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.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java b/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) {
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java b/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;
 	}
 
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/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");
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactoryUtils.java b/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();
 		}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/package-info.java b/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;
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/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()) {

Original file line number	Diff line number	Diff line change
`@@ -71,11 +71,11 @@ public DefaultKafkaConsumerFactory(Map<String, Object> configs,`
`71`	`71`	`this.valueDeserializer = valueDeserializer;`
`72`	`72`	`}`
`73`	`73`
`74`		`- public void setKeyDeserializer(Deserializer<K> keyDeserializer) {`
	`74`	`+ public void setKeyDeserializer(@Nullable Deserializer<K> keyDeserializer) {`
`75`	`75`	`this.keyDeserializer = keyDeserializer;`
`76`	`76`	`}`
`77`	`77`
`78`		`- public void setValueDeserializer(Deserializer<V> valueDeserializer) {`
	`78`	`+ public void setValueDeserializer(@Nullable Deserializer<V> valueDeserializer) {`
`79`	`79`	`this.valueDeserializer = valueDeserializer;`
`80`	`80`	`}`
`81`	`81`
`@@ -95,12 +95,14 @@ public Deserializer<V> getValueDeserializer() {`
`95`	`95`	`}`
`96`	`96`
`97`	`97`	`@Override`
`98`		`- public Consumer<K, V> createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) {`
	`98`	`+ public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,`
	`99`	`+ @Nullable String clientIdSuffix) {`
	`100`	`+`
`99`	`101`	`return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix);`
`100`	`102`	`}`
`101`	`103`
`102`		`- protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientIdPrefix,`
`103`		`- String clientIdSuffix) {`
	`104`	`+ protected KafkaConsumer<K, V> createKafkaConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,`
	`105`	`+ @Nullable String clientIdSuffix) {`
`104`	`106`
`105`	`107`	`boolean overrideClientIdPrefix = StringUtils.hasText(clientIdPrefix);`
`106`	`108`	`if (clientIdSuffix == null) {`
Original file line number	Diff line number	Diff line change
`@@ -116,11 +116,11 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,`
`116`	`116`	`this.valueSerializer = valueSerializer;`
`117`	`117`	`}`
`118`	`118`
`119`		`- public void setKeySerializer(Serializer<K> keySerializer) {`
	`119`	`+ public void setKeySerializer(@Nullable Serializer<K> keySerializer) {`
`120`	`120`	`this.keySerializer = keySerializer;`
`121`	`121`	`}`
`122`	`122`
`123`		`- public void setValueSerializer(Serializer<V> valueSerializer) {`
	`123`	`+ public void setValueSerializer(@Nullable Serializer<V> valueSerializer) {`
`124`	`124`	`this.valueSerializer = valueSerializer;`
`125`	`125`	`}`
`126`	`126`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`
`19`	`19`	`import org.apache.kafka.clients.producer.Producer;`
`20`	`20`
	`21`	`+import org.springframework.lang.Nullable;`
`21`	`22`	`import org.springframework.transaction.support.ResourceHolderSynchronization;`
`22`	`23`	`import org.springframework.transaction.support.TransactionSynchronization;`
`23`	`24`	`import org.springframework.transaction.support.TransactionSynchronizationManager;`
`@@ -72,7 +73,7 @@ public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(`
`72`	`73`	`return resourceHolder;`
`73`	`74`	`}`
`74`	`75`
`75`		`- public static <K, V> void releaseResources(KafkaResourceHolder<K, V> resourceHolder) {`
	`76`	`+ public static <K, V> void releaseResources(@Nullable KafkaResourceHolder<K, V> resourceHolder) {`
`76`	`77`	`if (resourceHolder != null) {`
`77`	`78`	`resourceHolder.getProducer().close();`
`78`	`79`	`}`