From eef31d4b342df1ab0a65d2ef825d050e847e9355 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 24 Feb 2021 11:34:29 -0500 Subject: [PATCH] Fix new Sonar smells; optimize some tests --- .../CachingClientConnectionFactoryTests.java | 18 +++++++++++++--- .../channel/SubscribableKafkaChannel.java | 10 +++++---- .../kafka/inbound/KafkaInboundGateway.java | 17 ++++++++------- .../KafkaMessageDrivenChannelAdapter.java | 21 ++++++++++--------- .../MessageSourceIntegrationTests.java | 5 ++++- 5 files changed, 45 insertions(+), 26 deletions(-) diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index 7339ab410a..bb3f749a4a 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -435,7 +435,7 @@ public class CachingClientConnectionFactoryTests { } @Test -// @Repeat(1000) // INT-3722 + // @Repeat(1000) // INT-3722 public void gatewayIntegrationTest() throws Exception { final List connectionIds = new ArrayList<>(); final AtomicBoolean okToRun = new AtomicBoolean(true); @@ -497,6 +497,7 @@ public class CachingClientConnectionFactoryTests { TcpConnection connection = cccf.getConnection(); await().atMost(Duration.ofSeconds(10)).until(() -> !connection.isOpen()); cccf.stop(); + cccf.destroy(); } @Test @@ -526,6 +527,9 @@ public class CachingClientConnectionFactoryTests { conn1 = cachingFactory.getConnection(); conn1.send(message); Mockito.verify(mockConn2).send(message); + conn1.close(); + cachingFactory.stop(); + cachingFactory.destroy(); } @Test @@ -592,6 +596,8 @@ public class CachingClientConnectionFactoryTests { assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue(); SimplePool pool = TestUtils.getPropertyValue(cachingFactory, "pool", SimplePool.class); assertThat(pool.getIdleCount()).isEqualTo(2); + cachingFactory.stop(); + cachingFactory.destroy(); server2.stop(); } @@ -650,6 +656,8 @@ public class CachingClientConnectionFactoryTests { assertThat(latch1.getCount()).isEqualTo(3); server1.stop(); server2.stop(); + cachingFactory.stop(); + cachingFactory.destroy(); } @Test //INT-3650 @@ -689,6 +697,7 @@ public class CachingClientConnectionFactoryTests { assertThat(connectionIds.get(101)).isSameAs(connectionIds.get(0)); in.stop(); cache.stop(); + cache.destroy(); } @SuppressWarnings("unchecked") @@ -721,11 +730,12 @@ public class CachingClientConnectionFactoryTests { TcpNetClientConnectionFactory out = new TcpNetClientConnectionFactory("localhost", port); out.setApplicationEventPublisher(mock(ApplicationEventPublisher.class)); CachingClientConnectionFactory cache = new CachingClientConnectionFactory(out, 2); - final TcpOutboundGateway gate = new TcpOutboundGateway(); + TcpOutboundGateway gate = new TcpOutboundGateway(); gate.setConnectionFactory(cache); QueueChannel outputChannel = new QueueChannel(); gate.setOutputChannel(outputChannel); gate.setBeanFactory(mock(BeanFactory.class)); + gate.setRemoteTimeout(20_000); gate.afterPropertiesSet(); LogAccessor logger = spy(TestUtils.getPropertyValue(gate, "logger", LogAccessor.class)); new DirectFieldAccessor(gate).setPropertyValue("logger", logger); @@ -761,6 +771,8 @@ public class CachingClientConnectionFactoryTests { handler.stop(); gate.stop(); verify(logger, never()).error(anyString()); + cache.stop(); + in.stop(); } @Test // INT-3728 diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java index 0dabc252ea..55f6020d45 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,6 +25,7 @@ import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.integration.support.management.ManageableSmartLifecycle; import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.core.KafkaOperations; +import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter; import org.springframework.kafka.support.Acknowledgment; @@ -110,9 +111,10 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su this.dispatcher = createDispatcher(); this.container = this.factory.createContainer(this.topic); String groupId = getGroupId(); - this.container.getContainerProperties().setGroupId(groupId != null ? groupId : getBeanName()); - this.container.getContainerProperties().setMessageListener( - new RecordMessagingMessageListenerAdapter(null, null) { + ContainerProperties containerProperties = this.container.getContainerProperties(); + containerProperties.setGroupId(groupId != null ? groupId : getBeanName()); + containerProperties.setMessageListener( + new RecordMessagingMessageListenerAdapter(null, null) { // NOSONAR - out of use @Override public void onMessage(ConsumerRecord record, Acknowledgment acknowledgment, diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index 2feefae122..b4b173987e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * Copyright 2018-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -38,6 +38,7 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.ConsumerSeekAware; +import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.MessageListener; import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; @@ -176,13 +177,13 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem super.onInit(); MessageListener kafkaListener = this.listener; if (this.retryTemplate != null) { - kafkaListener = new RetryingMessageListenerAdapter<>(kafkaListener, this.retryTemplate, - this.recoveryCallback); + kafkaListener = + new RetryingMessageListenerAdapter<>(kafkaListener, this.retryTemplate, this.recoveryCallback); this.retryTemplate.registerListener(this.listener); } - this.messageListenerContainer.getContainerProperties().setMessageListener(kafkaListener); - this.containerDeliveryAttemptPresent = this.messageListenerContainer.getContainerProperties() - .isDeliveryAttemptHeader(); + ContainerProperties containerProperties = this.messageListenerContainer.getContainerProperties(); + containerProperties.setMessageListener(kafkaListener); + this.containerDeliveryAttemptPresent = containerProperties.isDeliveryAttemptHeader(); } @Override @@ -266,7 +267,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem implements RetryListener { IntegrationRecordMessageListener() { - super(null, null); + super(null, null); // NOSONAR - out of use } @Override @@ -304,7 +305,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem } } else { - KafkaInboundGateway.this.logger.debug("Converter returned a null message for: " + record); + KafkaInboundGateway.this.logger.debug(() -> "Converter returned a null message for: " + record); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index 4ab767132c..7cee0d9982 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -38,6 +38,7 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.BatchMessageListener; import org.springframework.kafka.listener.ConsumerSeekAware; +import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.MessageListener; import org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.FilteringBatchMessageListenerAdapter; @@ -94,7 +95,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo private RetryTemplate retryTemplate; - private RecoveryCallback recoveryCallback; + private RecoveryCallback recoveryCallback; private boolean filterInRetry; @@ -210,7 +211,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * @param recoveryCallback the recovery callback. * @since 2.0.1 */ - public void setRecoveryCallback(RecoveryCallback recoveryCallback) { + public void setRecoveryCallback(RecoveryCallback recoveryCallback) { this.recoveryCallback = recoveryCallback; } @@ -280,6 +281,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo + "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to " + "send an error message when retries are exhausted"); } + ContainerProperties containerProperties = this.messageListenerContainer.getContainerProperties(); if (this.mode.equals(ListenerMode.record)) { MessageListener listener = this.recordListener; @@ -304,7 +306,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo this.ackDiscarded); } } - this.messageListenerContainer.getContainerProperties().setMessageListener(listener); + containerProperties.setMessageListener(listener); } else { BatchMessageListener listener = this.batchListener; @@ -313,10 +315,9 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo listener = new FilteringBatchMessageListenerAdapter<>(listener, this.recordFilterStrategy, this.ackDiscarded); } - this.messageListenerContainer.getContainerProperties().setMessageListener(listener); + containerProperties.setMessageListener(listener); } - this.containerDeliveryAttemptPresent = this.messageListenerContainer.getContainerProperties() - .isDeliveryAttemptHeader(); + this.containerDeliveryAttemptPresent = containerProperties.isDeliveryAttemptHeader(); } @Override @@ -402,7 +403,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo } } else { - KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: " + KafkaMessageDrivenChannelAdapter.this.logger.debug(() -> "Converter returned a null message for: " + kafkaConsumedObject); } } @@ -430,7 +431,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo implements RetryListener { IntegrationRecordMessageListener() { - super(null, null); + super(null, null); // NOSONAR - out of use } @Override @@ -520,7 +521,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo implements RetryListener { IntegrationBatchMessageListener() { - super(null, null); + super(null, null); // NOSONAR - out if use } @Override diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java index bd7b542b73..e3c3997d7c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -86,6 +86,9 @@ class MessageSourceIntegrationTests { } }); + + consumerProperties.setPollTimeout(10); + KafkaMessageSource source = new KafkaMessageSource<>(consumerFactory, consumerProperties); Map producerProps = KafkaTestUtils.producerProps(embeddedKafka);