Some code polishing for optimization

Author: izeye

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

@@ -128,8 +128,8 @@ protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientI
 			}
 			if (shouldModifyClientId) {
 				modifiedConfigs.put(ConsumerConfig.CLIENT_ID_CONFIG,
-					(overrideClientIdPrefix ? clientIdPrefix
-							: modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
+						(overrideClientIdPrefix ? clientIdPrefix
+								: modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
 			}
 			return createKafkaConsumer(modifiedConfigs);
 		}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java

@@ -23,7 +23,9 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
@@ -99,14 +101,11 @@ public List<KafkaMessageListenerContainer<K, V>> getContainers() {
 
 	@Override
 	public Collection<TopicPartition> getAssignedPartitions() {
-		List<TopicPartition> assigned = new ArrayList<>();
-		this.containers.forEach(c -> {
-			Collection<TopicPartition> assignedPartitions = c.getAssignedPartitions();
-			if (assignedPartitions != null) {
-				assigned.addAll(assignedPartitions);
-			}
-		});
-		return assigned;
+		return this.containers.stream()
+				.map(KafkaMessageListenerContainer::getAssignedPartitions)
+				.filter(Objects::nonNull)
+				.flatMap(assignedPartitions -> assignedPartitions.stream())
+				.collect(Collectors.toList());
 	}
 
 	@Override

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

@@ -497,7 +497,7 @@ else if (listener instanceof MessageListener) {
 			}
 			this.monitorTask = this.taskScheduler.scheduleAtFixedRate(() -> checkConsumer(),
 					this.containerProperties.getMonitorInterval() * 1000);
-			if (this.containerProperties.isLogContainerConfig() && this.logger.isInfoEnabled()) {
+			if (this.containerProperties.isLogContainerConfig()) {
 				this.logger.info(this);
 			}
 		}

Diff for: spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -225,7 +225,7 @@ public void testSimple() throws Exception {
 		assertThat(this.listener.listen4Consumer).isSameAs(KafkaTestUtils.getPropertyValue(KafkaTestUtils
 						.getPropertyValue(this.registry.getListenerContainer("qux"), "containers", List.class).get(0),
 				"listenerConsumer.consumer"));
-		assertThat(this.quxGroup.size()).isEqualTo(1);
+		assertThat(this.quxGroup).hasSize(1);
 		assertThat(this.quxGroup.get(0)).isSameAs(manualContainer);
 		List<?> containers = KafkaTestUtils.getPropertyValue(manualContainer, "containers", List.class);
 		assertThat(KafkaTestUtils.getPropertyValue(containers.get(0), "listenerConsumer.consumerGroupId"))

Diff for: spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java

@@ -68,7 +68,7 @@ public void testAddTopics() throws Exception {
 		this.admin.initialize();
 		topics = adminClient.describeTopics(Arrays.asList("foo", "bar"));
 		Map<String, TopicDescription> results = topics.all().get();
-		results.forEach((name, td) -> assertThat(td.partitions().size()).isEqualTo(name.equals("foo") ? 2 : 3));
+		results.forEach((name, td) -> assertThat(td.partitions()).hasSize(name.equals("foo") ? 2 : 3));
 		adminClient.close(10, TimeUnit.SECONDS);
 	}
 

Diff for: spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -145,7 +145,7 @@ public void testTemplate() throws Exception {
 		assertThat(metrics).isNotNull();
 		List<PartitionInfo> partitions = template.partitionsFor(INT_KEY_TOPIC);
 		assertThat(partitions).isNotNull();
-		assertThat(partitions.size()).isEqualTo(2);
+		assertThat(partitions).hasSize(2);
 		pf.destroy();
 	}
 
@@ -173,7 +173,7 @@ public void testTemplateWithTimestamps() throws Exception {
 		assertThat(metrics).isNotNull();
 		List<PartitionInfo> partitions = template.partitionsFor(INT_KEY_TOPIC);
 		assertThat(partitions).isNotNull();
-		assertThat(partitions.size()).isEqualTo(2);
+		assertThat(partitions).hasSize(2);
 		pf.destroy();
 	}
 

Diff for: spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java

@@ -119,7 +119,7 @@ public void testAutoCommit() throws Exception {
 		container.start();
 
 		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
-		assertThat(container.getAssignedPartitions().size()).isEqualTo(2);
+		assertThat(container.getAssignedPartitions()).hasSize(2);
 
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
@@ -137,7 +137,7 @@ public void testAutoCommit() throws Exception {
 		@SuppressWarnings("unchecked")
 		List<KafkaMessageListenerContainer<Integer, String>> containers = KafkaTestUtils.getPropertyValue(container,
 				"containers", List.class);
-		assertThat(containers.size()).isEqualTo(2);
+		assertThat(containers).hasSize(2);
 		for (int i = 0; i < 2; i++) {
 			assertThat(KafkaTestUtils.getPropertyValue(containers.get(i), "listenerConsumer.acks", Collection.class)
 					.size()).isEqualTo(0);
@@ -415,7 +415,7 @@ public ConsumerRecords<Integer, String> answer(InvocationOnMock invocation) thro
 		container.start();
 		List<KafkaMessageListenerContainer<Integer, String>> containers = KafkaTestUtils.getPropertyValue(container,
 				"containers", List.class);
-		assertThat(containers.size()).isEqualTo(3);
+		assertThat(containers).hasSize(3);
 		for (int i = 0; i < 3; i++) {
 			assertThat(KafkaTestUtils.getPropertyValue(containers.get(i), "topicPartitions",
 					TopicPartitionInitialOffset[].class).length).isEqualTo(i < 2 ? 2 : 3);

Diff for: spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -1071,7 +1071,7 @@ public void testDefinedPartitions() throws Exception {
 
 			@Override
 			public Consumer<Integer, String> createConsumer(String groupId, String clientIdPrefix,
-					String clientIdSufffix) {
+					String clientIdSuffix) {
 				return new KafkaConsumer<Integer, String>(props) {
 
 					@Override

Diff for: spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java

@@ -190,7 +190,7 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		ReplyingKafkaTemplate<Integer, String, String> template = new ReplyingKafkaTemplate<>(this.config.pf(), container);
 		template.start();
 		assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
-		assertThat(template.getAssignedReplyTopicPartitions().size()).isEqualTo(5);
+		assertThat(template.getAssignedReplyTopicPartitions()).hasSize(5);
 		assertThat(template.getAssignedReplyTopicPartitions().iterator().next().topic()).isEqualTo(topic);
 		return template;
 	}

Diff for: spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java

@@ -69,15 +69,15 @@ public void test() {
 		NonTrustedHeaderType ntht = (NonTrustedHeaderType) headers.get("fix");
 		assertThat(ntht.getHeaderValue()).isNotNull();
 		assertThat(ntht.getUntrustedType()).isEqualTo(Foo.class.getName());
-		assertThat(headers.size()).isEqualTo(6);
+		assertThat(headers).hasSize(6);
 		mapper.addTrustedPackages(getClass().getPackage().getName());
 		headers = new HashMap<>();
 		mapper.toHeaders(recordHeaders, headers);
 		assertThat(headers.get("foo")).isInstanceOf(byte[].class);
 		assertThat(new String((byte[]) headers.get("foo"))).isEqualTo("bar");
 		assertThat(headers.get("baz")).isEqualTo("qux");
 		assertThat(headers.get("fix")).isEqualTo(new Foo());
-		assertThat(headers.size()).isEqualTo(6);
+		assertThat(headers).hasSize(6);
 	}
 
 	public static final class Foo {

Diff for: spring-kafka/src/test/java/org/springframework/kafka/support/converter/BatchMessageConverterTests.java

@@ -53,9 +53,9 @@ public void testBatchConverters() throws Exception {
 		@SuppressWarnings("unchecked")
 		List<Map<String, Object>> converted = (List<Map<String, Object>>) headers
 				.get(KafkaHeaders.BATCH_CONVERTED_HEADERS);
-		assertThat(converted.size()).isEqualTo(3);
+		assertThat(converted).hasSize(3);
 		Map<String, Object> map = converted.get(0);
-		assertThat(map.size()).isEqualTo(1);
+		assertThat(map).hasSize(1);
 		assertThat(new String((byte[]) map.get("foo"))).isEqualTo("bar");
 	}
 
@@ -67,7 +67,7 @@ public void testNoMapper() {
 		MessageHeaders headers = testGuts(batchMessageConverter);
 		@SuppressWarnings("unchecked")
 		List<Headers> natives = (List<Headers>) headers.get(KafkaHeaders.NATIVE_HEADERS);
-		assertThat(natives.size()).isEqualTo(3);
+		assertThat(natives).hasSize(3);
 		Iterator<Header> iterator = natives.get(0).iterator();
 		assertThat(iterator.hasNext()).isEqualTo(true);
 		Header next = iterator.next();