Fix new Sonar issues

Author: garyrussell

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

@@ -64,6 +64,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	 */
 	public static final int DEFAULT_PHASE = Integer.MAX_VALUE - 100; // late phase
 
+	private static final int DEFAULT_TOPIC_CHECK_TIMEOUT = 30;
+
 	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
 
 	protected final ConsumerFactory<K, V> consumerFactory; // NOSONAR (final)
@@ -85,6 +87,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	private AfterRollbackProcessor<? super K, ? super V> afterRollbackProcessor =
 			new DefaultAfterRollbackProcessor<>();
 
+	private int topicCheckTimeout = DEFAULT_TOPIC_CHECK_TIMEOUT;
+
 	private volatile boolean running = false;
 
 	private volatile boolean paused;
@@ -265,6 +269,16 @@ public String getListenerId() {
 		return this.beanName; // the container factory sets the bean name to the id attribute
 	}
 
+	/**
+	 * How long to wait for {@link AdminClient#describeTopics(Collection)} result
+	 * futures to complete.
+	 * @param topicCheckTimeout the timeout in seconds; default 30.
+	 * @since 2.3
+	 */
+	public void setTopicCheckTimeout(int topicCheckTimeout) {
+		this.topicCheckTimeout = topicCheckTimeout;
+	}
+
 	@Override
 	public void setupMessageListener(Object messageListener) {
 		this.containerProperties.setMessageListener(messageListener);
@@ -304,7 +318,7 @@ protected void checkTopics() {
 						.stream()
 						.filter(entry -> {
 							try {
-								entry.getValue().get(30, TimeUnit.SECONDS);
+								entry.getValue().get(this.topicCheckTimeout, TimeUnit.SECONDS);
 								return false;
 							}
 							catch (@SuppressWarnings("unused") Exception e) {

spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java

@@ -29,7 +29,6 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Header;
@@ -58,24 +57,26 @@ public class AggregatingReplyingKafkaTemplate<K, V, R>
 		implements BatchConsumerAwareMessageListener<K, Collection<ConsumerRecord<K, R>>> {
 
 	/**
-	 * Pseudo topic name for the "outer" {@link ConsumerRecords} that has the aggregated
+	 * Pseudo topic name for the "outer" {@link ConsumerRecord} that has the aggregated
 	 * results in its value after a normal release by the release strategy.
 	 */
 	public static final String AGGREGATED_RESULTS_TOPIC = "aggregatedResults";
 
 	/**
-	 * Pseudo topic name for the "outer" {@link ConsumerRecords} that has the aggregated
+	 * Pseudo topic name for the "outer" {@link ConsumerRecord} that has the aggregated
 	 * results in its value after a timeout.
 	 */
 	public static final String PARTIAL_RESULTS_AFTER_TIMEOUT_TOPIC = "partialResultsAfterTimeout";
 
+	private static final int DEFAULT_COMMIT_TIMEOUT = 30;
+
 	private final Map<CorrelationKey, Set<RecordHolder<K, R>>> pending = new HashMap<>();
 
 	private final Map<TopicPartition, Long> offsets = new HashMap<>();
 
 	private final Predicate<Collection<ConsumerRecord<K, R>>> releaseStrategy;
 
-	private Duration commitTimeout = Duration.ofSeconds(30);
+	private Duration commitTimeout = Duration.ofSeconds(DEFAULT_COMMIT_TIMEOUT);
 
 	private boolean returnPartialOnTimeout;
 
@@ -111,7 +112,7 @@ public void setCommitTimeout(Duration commitTimeout) {
 	 * Set to true to return a partial result when a request times out.
 	 * @param returnPartialOnTimeout true to return a partial result.
 	 */
-	public void setReturnPartialOnTimeout(boolean returnPartialOnTimeout) {
+	public synchronized void setReturnPartialOnTimeout(boolean returnPartialOnTimeout) {
 		this.returnPartialOnTimeout = returnPartialOnTimeout;
 	}