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

Fix new Sonar smells in ContentEnricher

Browse Source 
* Increase timeouts in sporadically failing correlation handler tests
This commit is contained in:
Artem Bilan
2021-04-21 09:59:00 -04:00
parent 700ec4ed31
commit fc941e6a19
3 changed files with 83 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java
View File

@@ -292,7 +292,8 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem
this.targetEvaluationContext = targetContext;
if (beanFactory != null) {
configureHeaderExpressions(beanFactory);
configureHeaderExpressions(beanFactory, this.headerExpressions);
configureHeaderExpressions(beanFactory, this.nullResultHeaderExpressions);
}
}
@@ -316,29 +317,18 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem
}
}
private void configureHeaderExpressions(BeanFactory beanFactory) {
private void configureHeaderExpressions(BeanFactory beanFactory,
Map<String, HeaderValueMessageProcessor<?>> headerExpressions) {
boolean checkReadOnlyHeaders = getMessageBuilderFactory() instanceof DefaultMessageBuilderFactory;
for (Map.Entry<String, HeaderValueMessageProcessor<?>> entry : this.headerExpressions.entrySet()) {
for (Map.Entry<String, HeaderValueMessageProcessor<?>> entry : headerExpressions.entrySet()) {
if (checkReadOnlyHeaders &&
(MessageHeaders.ID.equals(entry.getKey()) || MessageHeaders.TIMESTAMP.equals(entry.getKey()))) {
throw new BeanInitializationException(
"ContentEnricher cannot override 'id' and 'timestamp' read-only headers.\n" +
"Wrong 'headerExpressions' [" + this.headerExpressions
+ "] configuration for " + getComponentName());
}
if (entry.getValue() instanceof BeanFactoryAware) {
((BeanFactoryAware) entry.getValue()).setBeanFactory(beanFactory);
}
}
for (Map.Entry<String, HeaderValueMessageProcessor<?>> entry :
this.nullResultHeaderExpressions.entrySet()) {
if (checkReadOnlyHeaders &&
(MessageHeaders.ID.equals(entry.getKey()) || MessageHeaders.TIMESTAMP.equals(entry.getKey()))) {
throw new BeanInitializationException(
"ContentEnricher cannot override 'id' and 'timestamp' read-only headers.\n" +
"Wrong 'nullResultHeaderExpressions' [" + this.nullResultHeaderExpressions
"Wrong 'headerExpressions' [" + headerExpressions
+ "] configuration for " + getComponentName());
}
if (entry.getValue() instanceof BeanFactoryAware) {

133
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 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.
@@ -17,13 +17,14 @@
package org.springframework.integration.aggregator;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
@@ -53,7 +54,7 @@ public class ConcurrentAggregatorTests {
private final MessageGroupStore store = new SimpleMessageStore();
@Before
@BeforeEach
public void configureAggregator() {
this.taskExecutor = new SimpleAsyncTaskExecutor();
this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), this.store);
@@ -68,12 +69,9 @@ public class ConcurrentAggregatorTests {
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, null);
CountDownLatch latch = new CountDownLatch(3);
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message1, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message2, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message3, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch));
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -84,16 +82,13 @@ public class ConcurrentAggregatorTests {
}
@Test
@Ignore
@Disabled
// dropped backwards compatibility for duplicate ID's
public void testCompleteGroupWithinTimeoutWithSameId() {
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel,
"ID#1");
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel,
"ID#1");
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel,
"ID#1");
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, "ID#1");
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, "ID#1");
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, "ID#1");
CountDownLatch latch = new CountDownLatch(3);
// for testing the duplication scenario, the messages must be processed
// synchronously
@@ -107,20 +102,17 @@ public class ConcurrentAggregatorTests {
@Test
public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException {
QueueChannel discardChannel = new QueueChannel();
this.aggregator.setDiscardChannel(discardChannel);
QueueChannel replyChannel = new QueueChannel();
Message<?> message = createMessage(3, "ABC", 2, 1, replyChannel, null);
CountDownLatch latch = new CountDownLatch(1);
AggregatorTestTask task = new AggregatorTestTask(this.aggregator,
message, latch);
AggregatorTestTask task = new AggregatorTestTask(this.aggregator, message, latch);
this.taskExecutor.execute(task);
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latch
.getCount()).as("Task should have completed within timeout").isEqualTo(0);
assertThat(latch.getCount()).as("Task should have completed within timeout").isEqualTo(0);
Message<?> reply = replyChannel.receive(10);
assertThat(reply).as("No message should have been sent normally").isNull();
this.store.expireMessageGroups(-10000);
@@ -136,10 +128,8 @@ public class ConcurrentAggregatorTests {
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
CountDownLatch latch = new CountDownLatch(2);
AggregatorTestTask task1 = new AggregatorTestTask(this.aggregator,
message1, latch);
AggregatorTestTask task2 = new AggregatorTestTask(this.aggregator,
message2, latch);
AggregatorTestTask task1 = new AggregatorTestTask(this.aggregator, message1, latch);
AggregatorTestTask task2 = new AggregatorTestTask(this.aggregator, message2, latch);
this.taskExecutor.execute(task1);
this.taskExecutor.execute(task2);
@@ -161,25 +151,16 @@ public class ConcurrentAggregatorTests {
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel1, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel1, null);
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel1, null);
Message<?> message4 = createMessage(11, "XYZ", 3, 1, replyChannel2,
null);
Message<?> message5 = createMessage(13, "XYZ", 3, 2, replyChannel2,
null);
Message<?> message6 = createMessage(17, "XYZ", 3, 3, replyChannel2,
null);
Message<?> message4 = createMessage(11, "XYZ", 3, 1, replyChannel2, null);
Message<?> message5 = createMessage(13, "XYZ", 3, 2, replyChannel2, null);
Message<?> message6 = createMessage(17, "XYZ", 3, 3, replyChannel2, null);
CountDownLatch latch = new CountDownLatch(6);
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message1, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message6, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message2, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message5, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message3, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message4, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message6, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message5, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch));
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -194,7 +175,7 @@ public class ConcurrentAggregatorTests {
}
@Test
@Ignore
@Disabled
// dropped backwards compatibility for setting capacity limit (it's always
// Integer.MAX_VALUE)
public void testTrackedCorrelationIdsCapacityAtLimit() {
@@ -202,23 +183,19 @@ public class ConcurrentAggregatorTests {
QueueChannel discardChannel = new QueueChannel();
// this.aggregator.setTrackedCorrelationIdCapacity(3);
this.aggregator.setDiscardChannel(discardChannel);
this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null));
assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(1);
this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel, null));
assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(3);
this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel, null));
assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(4);
// next message with same correlation ID is discarded
this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, null));
assertThat(discardChannel.receive(1000).getPayload()).isEqualTo(2);
}
@Test
@Ignore
@Disabled
// dropped backwards compatibility for setting capacity limit (it's always
// Integer.MAX_VALUE)
public void testTrackedCorrelationIdsCapacityPassesLimit() {
@@ -226,29 +203,24 @@ public class ConcurrentAggregatorTests {
QueueChannel discardChannel = new QueueChannel();
// this.aggregator.setTrackedCorrelationIdCapacity(3);
this.aggregator.setDiscardChannel(discardChannel);
this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null));
assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(1);
this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel, null));
assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(2);
this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel, null));
assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(3);
this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel, null));
assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(4);
this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel,
null));
this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel, null));
assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(5);
assertThat(discardChannel.receive(0)).isNull();
}
@Test(expected = MessageHandlingException.class)
@Test
public void testExceptionThrownIfNoCorrelationId() {
Message<?> message = createMessage(3, null, 2, 1, new QueueChannel(),
null);
this.aggregator.handleMessage(message);
Message<?> message = createMessage(3, null, 2, 1, new QueueChannel(), null);
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.aggregator.handleMessage(message));
}
@Test
@@ -262,16 +234,12 @@ public class ConcurrentAggregatorTests {
this.aggregator.setReleaseStrategy(new SequenceSizeReleaseStrategy());
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message1, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message2, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message3, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message4, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch));
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latch.await(20, TimeUnit.SECONDS)).isTrue();
Message<?> reply = replyChannel.receive(10000);
assertThat(reply).as("A message should be aggregated").isNotNull();
@@ -283,10 +251,11 @@ public class ConcurrentAggregatorTests {
Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel replyChannel, String predefinedId) {
MessageBuilder<Object> builder = MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId).setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(replyChannel);
MessageBuilder<Object> builder =
MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId).setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(replyChannel);
if (predefinedId != null) {
builder.setHeader(MessageHeaders.ID, predefinedId);
}

45
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 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.
@@ -24,8 +24,8 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -56,7 +56,7 @@ public class ResequencerTests {
private final MessageGroupStore store = new SimpleMessageStore();
@Before
@BeforeEach
public void configureResequencer() {
this.resequencer = new ResequencingMessageHandler(processor, store, null, null);
this.resequencer.setBeanFactory(mock(BeanFactory.class));
@@ -64,7 +64,7 @@ public class ResequencerTests {
}
@Test
public void testBasicResequencing() throws InterruptedException {
public void testBasicResequencing() {
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 3, 3, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 3, 1, replyChannel);
@@ -84,7 +84,7 @@ public class ResequencerTests {
}
@Test
public void testBasicResequencingA() throws InterruptedException {
public void testBasicResequencingA() {
SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();
releaseStrategy.setReleasePartialSequences(true);
this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy);
@@ -103,7 +103,7 @@ public class ResequencerTests {
}
@Test
public void testBasicUnboundedResequencing() throws InterruptedException {
public void testBasicUnboundedResequencing() {
SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();
releaseStrategy.setReleasePartialSequences(true);
this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy);
@@ -158,7 +158,7 @@ public class ResequencerTests {
}
@Test
public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException {
public void testResequencingWithIncompleteSequenceRelease() {
this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true));
// INT-3846
this.resequencer.setMessageStore(new SimpleMessageStore(3));
@@ -190,7 +190,7 @@ public class ResequencerTests {
}
@Test
public void testResequencingWithCapacity() throws InterruptedException {
public void testResequencingWithCapacity() {
this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true));
// INT-3846
this.resequencer.setMessageStore(new SimpleMessageStore(3, 2));
@@ -210,7 +210,7 @@ public class ResequencerTests {
}
@Test
public void testResequencingWithPartialSequenceAndComparator() throws InterruptedException {
public void testResequencingWithPartialSequenceAndComparator() {
this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true));
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("456", "ABC", 4, 2, replyChannel);
@@ -240,7 +240,7 @@ public class ResequencerTests {
}
@Test
public void testResequencingWithDiscard() throws InterruptedException {
public void testResequencingWithDiscard() {
QueueChannel discardChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 4, 2, null);
Message<?> message2 = createMessage("456", "ABC", 4, 1, null);
@@ -269,7 +269,7 @@ public class ResequencerTests {
}
@Test
public void testResequencingWithDifferentSequenceSizes() throws InterruptedException {
public void testResequencingWithDifferentSequenceSizes() {
QueueChannel discardChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 4, 2, null);
Message<?> message2 = createMessage("456", "ABC", 5, 1, null);
@@ -288,7 +288,7 @@ public class ResequencerTests {
}
@Test
public void testResequencingWithWrongSequenceSizeAndNumber() throws InterruptedException {
public void testResequencingWithWrongSequenceSizeAndNumber() {
QueueChannel discardChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 2, 4, null);
this.resequencer.setSendPartialResultOnExpiry(false);
@@ -301,7 +301,7 @@ public class ResequencerTests {
}
@Test
public void testResequencingWithCompleteSequenceRelease() throws InterruptedException {
public void testResequencingWithCompleteSequenceRelease() {
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
@@ -343,7 +343,7 @@ public class ResequencerTests {
}
@Test
public void testTimeoutDefaultExpiry() throws InterruptedException {
public void testTimeoutDefaultExpiry() {
this.resequencer.setGroupTimeoutExpression(new SpelExpressionParser().parseExpression("100"));
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
@@ -359,7 +359,7 @@ public class ResequencerTests {
this.resequencer.handleMessage(message2);
Message<?> out1 = replyChannel.receive(10);
assertThat(out1).isNull();
out1 = discardChannel.receive(10000);
out1 = discardChannel.receive(20000);
assertThat(out1).isNotNull();
Message<?> out2 = discardChannel.receive(10);
assertThat(out2).isNotNull();
@@ -370,7 +370,7 @@ public class ResequencerTests {
}
@Test
public void testTimeoutDontExpire() throws InterruptedException {
public void testTimeoutDontExpire() {
this.resequencer.setGroupTimeoutExpression(new SpelExpressionParser().parseExpression("100"));
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
@@ -387,7 +387,7 @@ public class ResequencerTests {
this.resequencer.handleMessage(message2);
Message<?> out1 = replyChannel.receive(0);
assertThat(out1).isNull();
out1 = discardChannel.receive(10_000);
out1 = discardChannel.receive(20_000);
assertThat(out1).isNotNull();
Message<?> out2 = discardChannel.receive(10_000);
assertThat(out2).isNotNull();
@@ -401,8 +401,13 @@ public class ResequencerTests {
private static Message<?> createMessage(String payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel replyChannel) {
return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber).setReplyChannel(replyChannel).build();
return MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId)
.setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(replyChannel)
.build();
}
}