Use isEmpty() Instead of size(); toList()

Sonar:

Collection.isEmpty() should be used to test for emptiness (java:S1155)

CODE_SMELL_MINOR

Using Collection.size() to test for emptiness works, but using Collection.isEmpty() makes the code more readable and can be more performant. The time complexity of any isEmpty() method implementation should be O(1) whereas some implementations of size() can be O(n).

Also

"Stream.toList()" method should be used instead of "collectors" when unmodifiable list needed (java:S6204)

CODE_SMELL_MAJOR

In Java 8 Streams were introduced to support chaining of operations over collections
in a functional style. The most common way to save a result of such chains is to save
 them to some collection (usually List). To do so there is a terminal method collect
 that can be used with a library of Collectors. The key problem is that
`.collect(Collectors.toList())` actually returns a mutable kind of List
while in the majority of cases unmodifiable lists are preferred.
In Java 10 a new collector appeared to return an unmodifiable list:
`toUnmodifiableList()`. This does the trick but results in verbose code.
 Since Java 16 there is now a better variant to produce an unmodifiable list
 directly from a stream: Stream.toList().

Author: garyrussell

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

@@ -1428,7 +1428,7 @@ protected void pollAndInvoke() {
 			doProcessCommits();
 			fixTxOffsetsIfNeeded();
 			idleBetweenPollIfNecessary();
-			if (this.seeks.size() > 0) {
+			if (!this.seeks.isEmpty()) {
 				processSeeks();
 			}
 			pauseConsumerIfNecessary();
@@ -1586,7 +1586,7 @@ private void fixTxOffsetsIfNeeded() {
 							toFix.put(tp, createOffsetAndMetadata(position));
 						}
 					});
-					if (toFix.size() > 0) {
+					if (!toFix.isEmpty()) {
 						this.logger.debug(() -> "Fixing TX offsets: " + toFix);
 						if (this.kafkaTxManager == null) {
 							if (this.syncCommits) {
@@ -1689,7 +1689,7 @@ private synchronized void captureOffsets(ConsumerRecords<K, V> records) {
 		}
 
 		private void checkRebalanceCommits() {
-			if (this.commitsDuringRebalance.size() > 0) {
+			if (!this.commitsDuringRebalance.isEmpty()) {
 				// Attempt to recommit the offsets for partitions that we still own
 				Map<TopicPartition, OffsetAndMetadata> commits = this.commitsDuringRebalance.entrySet()
 						.stream()
@@ -1729,7 +1729,8 @@ private void debugRecords(@Nullable ConsumerRecords<K, V> records) {
 							.flatMap(p -> records.records(p).stream())
 							// map to same format as send metadata toString()
 							.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
-							.collect(Collectors.toList()).toString());
+							.toList()
+							.toString());
 				}
 			}
 		}
@@ -1756,11 +1757,11 @@ private void pauseConsumerIfNecessary() {
 		}
 
 		private void doPauseConsumerIfNecessary() {
-			if (this.pausedForNack.size() > 0) {
+			if (!this.pausedForNack.isEmpty()) {
 				this.logger.debug("Still paused for nack sleep");
 				return;
 			}
-			if (this.offsetsInThisBatch != null && this.offsetsInThisBatch.size() > 0 && !this.pausedForAsyncAcks) {
+			if (this.offsetsInThisBatch != null && !this.offsetsInThisBatch.isEmpty() && !this.pausedForAsyncAcks) {
 				this.pausedForAsyncAcks = true;
 				this.logger.debug(() -> "Pausing for incomplete async acks: " + this.offsetsInThisBatch);
 			}
@@ -1801,7 +1802,7 @@ else if (this.offsetsInThisBatch != null) {
 		}
 
 		private void doResumeConsumerIfNeccessary() {
-			if (this.pausedForAsyncAcks && this.offsetsInThisBatch.size() == 0) {
+			if (this.pausedForAsyncAcks && this.offsetsInThisBatch.isEmpty()) {
 				this.pausedForAsyncAcks = false;
 				this.logger.debug("Resuming after manual async acks cleared");
 			}
@@ -1823,8 +1824,8 @@ private void pausePartitionsIfNecessary() {
 						.stream()
 						.filter(tp -> isPartitionPauseRequested(tp)
 								&& !pausedConsumerPartitions.contains(tp))
-						.collect(Collectors.toList());
-				if (partitionsToPause.size() > 0) {
+						.toList();
+				if (!partitionsToPause.isEmpty()) {
 					this.consumer.pause(partitionsToPause);
 					this.pausedPartitions.addAll(partitionsToPause);
 					this.logger.debug(() -> "Paused consumption from " + partitionsToPause);
@@ -1840,8 +1841,8 @@ private void resumePartitionsIfNecessary() {
 						.stream()
 						.filter(tp -> !isPartitionPauseRequested(tp)
 								&& this.pausedPartitions.contains(tp))
-						.collect(Collectors.toList());
-				if (partitionsToResume.size() > 0) {
+						.toList();
+				if (!partitionsToResume.isEmpty()) {
 					this.consumer.resume(partitionsToResume);
 					this.pausedPartitions.removeAll(partitionsToResume);
 					this.logger.debug(() -> "Resumed consumption from " + partitionsToResume);
@@ -1877,7 +1878,7 @@ private void checkIdle() {
 		private void idleBetweenPollIfNecessary() {
 			long idleBetweenPolls = this.containerProperties.getIdleBetweenPolls();
 			Collection<TopicPartition> assigned = getAssignedPartitions();
-			if (idleBetweenPolls > 0 && assigned != null && assigned.size() > 0) {
+			if (idleBetweenPolls > 0 && assigned != null && !assigned.isEmpty()) {
 				idleBetweenPolls = Math.min(idleBetweenPolls,
 						this.maxPollInterval - (System.currentTimeMillis() - this.lastPoll)
 								- 5000); // NOSONAR - less by five seconds to avoid race condition with rebalance
@@ -1960,7 +1961,7 @@ protected void handleConsumerException(Exception e) {
 
 		private void commitPendingAcks() {
 			processCommits();
-			if (this.offsets.size() > 0) {
+			if (!this.offsets.isEmpty()) {
 				// we always commit after stopping the invoker
 				commitIfNecessary();
 			}
@@ -2056,19 +2057,19 @@ private synchronized void ackInOrder(ConsumerRecord<K, V> record) {
 			TopicPartition part = new TopicPartition(record.topic(), record.partition());
 			List<Long> offs = this.offsetsInThisBatch.get(part);
 			List<ConsumerRecord<K, V>> deferred = this.deferredOffsets.get(part);
-			if (offs.size() > 0) {
+			if (!offs.isEmpty()) {
 				if (offs.get(0) == record.offset()) {
 					offs.remove(0);
 					ConsumerRecord<K, V> recordToAck = record;
-					if (deferred.size() > 0) {
+					if (!deferred.isEmpty()) {
 						Collections.sort(deferred, (a, b) -> Long.compare(a.offset(), b.offset()));
-						while (deferred.size() > 0 && deferred.get(0).offset() == recordToAck.offset() + 1) {
+						while (!deferred.isEmpty() && deferred.get(0).offset() == recordToAck.offset() + 1) {
 							recordToAck = deferred.remove(0);
 							offs.remove(0);
 						}
 					}
 					processAck(recordToAck);
-					if (offs.size() == 0) {
+					if (offs.isEmpty()) {
 						this.deferredOffsets.remove(part);
 						this.offsetsInThisBatch.remove(part);
 					}
@@ -2148,7 +2149,7 @@ private void invokeBatchListener(final ConsumerRecords<K, V> recordsArg) {
 			if (!this.wantsFullRecords) {
 				recordList = createRecordList(records);
 			}
-			if (this.wantsFullRecords || recordList.size() > 0) {
+			if (this.wantsFullRecords || !recordList.isEmpty()) {
 				if (this.transactionTemplate != null) {
 					invokeBatchListenerInTx(records, recordList); // NOSONAR
 				}
@@ -2625,7 +2626,7 @@ private boolean checkImmediatePause(Iterator<ConsumerRecord<K, V>> iterator) {
 					remaining.computeIfAbsent(new TopicPartition(next.topic(), next.partition()),
 							tp -> new ArrayList<ConsumerRecord<K, V>>()).add(next);
 				}
-				if (remaining.size() > 0) {
+				if (!remaining.isEmpty()) {
 					this.remainingRecords = new ConsumerRecords<>(remaining);
 					return true;
 				}
@@ -2693,7 +2694,7 @@ private void handleNack(final ConsumerRecords<K, V> records, final ConsumerRecor
 			Iterator<ConsumerRecord<K, V>> iterator2 = records.iterator();
 			while (iterator2.hasNext()) {
 				ConsumerRecord<K, V> next = iterator2.next();
-				if (list.size() > 0 || recordsEqual(record, next)) {
+				if (!list.isEmpty() || recordsEqual(record, next)) {
 					list.add(next);
 				}
 			}
@@ -2908,7 +2909,7 @@ private void invokeErrorHandler(final ConsumerRecord<K, V> record,
 					records.computeIfAbsent(new TopicPartition(next.topic(), next.partition()),
 							tp -> new ArrayList<ConsumerRecord<K, V>>()).add(next);
 				}
-				if (records.size() > 0) {
+				if (!records.isEmpty()) {
 					this.remainingRecords = new ConsumerRecords<>(records);
 					this.pauseForPending = true;
 				}
@@ -3180,10 +3181,10 @@ private void initPartitionsIfNeeded() {
 		private void doInitialSeeks(Map<TopicPartition, OffsetMetadata> partitions, Set<TopicPartition> beginnings,
 				Set<TopicPartition> ends) {
 
-			if (beginnings.size() > 0) {
+			if (!beginnings.isEmpty()) {
 				this.consumer.seekToBeginning(beginnings);
 			}
-			if (ends.size() > 0) {
+			if (!ends.isEmpty()) {
 				this.consumer.seekToEnd(ends);
 			}
 			for (Entry<TopicPartition, OffsetMetadata> entry : partitions.entrySet()) {
@@ -3312,7 +3313,7 @@ public void seekToBeginning(String topic, int partition) {
 		public void seekToBeginning(Collection<TopicPartition> partitions) {
 			this.seeks.addAll(partitions.stream()
 					.map(tp -> new TopicPartitionOffset(tp.topic(), tp.partition(), SeekPosition.BEGINNING))
-					.collect(Collectors.toList()));
+					.toList());
 		}
 
 		@Override
@@ -3324,7 +3325,7 @@ public void seekToEnd(String topic, int partition) {
 		public void seekToEnd(Collection<TopicPartition> partitions) {
 			this.seeks.addAll(partitions.stream()
 					.map(tp -> new TopicPartitionOffset(tp.topic(), tp.partition(), SeekPosition.END))
-					.collect(Collectors.toList()));
+					.toList());
 		}
 
 		@Override
@@ -3564,15 +3565,15 @@ private void repauseIfNeeded(Collection<TopicPartition> partitions) {
 						toRepause.add(tp);
 					}
 				});
-				if (!ListenerConsumer.this.consumerPaused && toRepause.size() > 0) {
+				if (!ListenerConsumer.this.consumerPaused && !toRepause.isEmpty()) {
 					ListenerConsumer.this.consumer.pause(toRepause);
 					ListenerConsumer.this.logger.debug(() -> "Paused consumption from: " + toRepause);
 					publishConsumerPausedEvent(toRepause, "Re-paused after rebalance");
 				}
 				this.revoked.removeAll(toRepause);
 				ListenerConsumer.this.pausedPartitions.removeAll(this.revoked);
 				this.revoked.clear();
-				if (ListenerConsumer.this.pausedForNack.size() > 0) {
+				if (!ListenerConsumer.this.pausedForNack.isEmpty()) {
 					ListenerConsumer.this.consumer.pause(ListenerConsumer.this.pausedForNack);
 				}
 			}
@@ -3597,7 +3598,7 @@ private boolean collectAndCommitIfNecessary(Collection<TopicPartition> partition
 						return false;
 					}
 				}
-				if (offsetsToCommit.size() > 0) {
+				if (!offsetsToCommit.isEmpty()) {
 					commitCurrentOffsets(offsetsToCommit);
 				}
 				return true;