Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix new Sonar smells; optimize some tests

Browse Source
This commit is contained in:
Artem Bilan
2021-02-24 11:34:29 -05:00
parent a0093f016d
commit eef31d4b34
5 changed files with 45 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
View File

@@ -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<String> 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

10
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java
View File

@@ -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<Object, Object>(null, null) {
ContainerProperties containerProperties = this.container.getContainerProperties();
containerProperties.setGroupId(groupId != null ? groupId : getBeanName());
containerProperties.setMessageListener(
new RecordMessagingMessageListenerAdapter<Object, Object>(null, null) { // NOSONAR - out of use
@Override
public void onMessage(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment,

17
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
View File

@@ -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<K, V, R> extends MessagingGatewaySupport implem
super.onInit();
MessageListener<K, V> 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<K, V, R> extends MessagingGatewaySupport implem
implements RetryListener {
IntegrationRecordMessageListener() {
super(null, null);
super(null, null); // NOSONAR - out of use
}
@Override
@@ -304,7 +305,7 @@ public class KafkaInboundGateway<K, V, R> 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);
}
}

21
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -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<K, V> extends MessageProducerSuppo
private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback;
private RecoveryCallback<?> recoveryCallback;
private boolean filterInRetry;
@@ -210,7 +211,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
* @param recoveryCallback the recovery callback.
* @since 2.0.1
*/
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
public void setRecoveryCallback(RecoveryCallback<?> recoveryCallback) {
this.recoveryCallback = recoveryCallback;
}
@@ -280,6 +281,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> 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<K, V> listener = this.recordListener;
@@ -304,7 +306,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
this.ackDiscarded);
}
}
this.messageListenerContainer.getContainerProperties().setMessageListener(listener);
containerProperties.setMessageListener(listener);
}
else {
BatchMessageListener<K, V> listener = this.batchListener;
@@ -313,10 +315,9 @@ public class KafkaMessageDrivenChannelAdapter<K, V> 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<K, V> 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<K, V> extends MessageProducerSuppo
implements RetryListener {
IntegrationRecordMessageListener() {
super(null, null);
super(null, null); // NOSONAR - out of use
}
@Override
@@ -520,7 +521,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
implements RetryListener {
IntegrationBatchMessageListener() {
super(null, null);
super(null, null); // NOSONAR - out if use
}
@Override

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java
View File

@@ -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<Integer, String> source = new KafkaMessageSource<>(consumerFactory, consumerProperties);
Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);