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

Fix race condition in the AMQP ChannelTests

Browse Source 
https://build.spring.io/browse/INT-AT42SIO-502

There is a small time window when we remove the current consumer
from the local store, but there is no a new one yet.
So, we have to check the `Set`(`Map`) size before calling its `iterator`

**Cherry-pick to 4.3.x**

Conflicts:
	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java
This commit is contained in:
Artem Bilan
2017-05-08 10:28:56 -04:00
parent f166bd1bd7
commit 0a7dfa96e7
1 changed files with 15 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java
View File

@@ -25,6 +25,7 @@ import static org.mockito.Mockito.mock;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
@@ -110,19 +111,24 @@ public class ChannelTests extends LogAdjustingTestSupport {
assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size());
}
@SuppressWarnings("unchecked")
private void waitForNewConsumer(PublishSubscribeAmqpChannel channel, BlockingQueueConsumer consumer)
throws Exception {
BlockingQueueConsumer newConsumer = (BlockingQueueConsumer) TestUtils
.getPropertyValue(channel, "container.consumers", Map.class).keySet().iterator().next();
final Object consumersMonitor = TestUtils.getPropertyValue(channel, "container.consumersMonitor");
int n = 0;
boolean newConsumerIsConsuming = newConsumer != consumer && TestUtils.getPropertyValue(newConsumer,
"consumerTags", Map.class).size() > 0;
while (n++ < 100 && !newConsumerIsConsuming) {
while (n++ < 100) {
Set<BlockingQueueConsumer> consumers = TestUtils.getPropertyValue(channel, "container.consumers", Map.class).keySet();
synchronized (consumersMonitor) {
if (!consumers.isEmpty()) {
BlockingQueueConsumer newConsumer = consumers.iterator().next();
if (newConsumer != consumer && TestUtils.getPropertyValue(newConsumer,
"consumerTags", Map.class).size() > 0) {
break;
}
}
}
Thread.sleep(100);
newConsumer = (BlockingQueueConsumer) TestUtils
.getPropertyValue(channel, "container.consumers", Map.class).keySet().iterator().next();
newConsumerIsConsuming = newConsumer != consumer && TestUtils.getPropertyValue(newConsumer,
"consumerTags", Map.class).size() > 0;
}
assertTrue("Failed to restart consumer", n < 100);
}