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

Increase timeout for PollingConsumerEndpointTests

Browse Source 
https://build.spring.io/browse/INT-SI50X-20

Looks like 5 seconds is not enough to wait for the test completion,
especially on CI server

* Refactor `PollingConsumerEndpointTests` to use `OnlyOnceTrigger`
from the test-support module instead of local `Trigger` implementation
* Increase timeout for latch in the `OnlyOnceTrigger` to 10 seconds

**Cherry-pick to 5.0.x**
This commit is contained in:
Artem Bilan
2018-04-02 14:44:56 -04:00
parent 5cdcd81444
commit a65d05cdbf
2 changed files with 81 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2017 the original author or authors. * Copyright 2002-2018 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -19,10 +19,6 @@ package org.springframework.integration.endpoint;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import org.junit.After; import org.junit.After;
@@ -33,12 +29,11 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.MessageRejectedException; import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.support.MessagingExceptionWrapper; import org.springframework.integration.support.MessagingExceptionWrapper;
import org.springframework.integration.test.util.OnlyOnceTrigger;
import org.springframework.messaging.Message; import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel; import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ErrorHandler; import org.springframework.util.ErrorHandler;
@@ -46,131 +41,132 @@ import org.springframework.util.ErrorHandler;
* @author Iwein Fuld * @author Iwein Fuld
* @author Mark Fisher * @author Mark Fisher
* @author Kiel Boatman * @author Kiel Boatman
* @author Artem Bilan
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings({ "rawtypes", "unchecked" })
public class PollingConsumerEndpointTests { public class PollingConsumerEndpointTests {
private PollingConsumer endpoint; private final OnlyOnceTrigger trigger = new OnlyOnceTrigger();
private final TestTrigger trigger = new TestTrigger();
private final TestConsumer consumer = new TestConsumer(); private final TestConsumer consumer = new TestConsumer();
@SuppressWarnings("rawtypes") private final Message message = new GenericMessage<>("test");
private final Message message = new GenericMessage<String>("test");
@SuppressWarnings("rawtypes") private final Message badMessage = new GenericMessage<>("bad");
private final Message badMessage = new GenericMessage<String>("bad");
private final TestErrorHandler errorHandler = new TestErrorHandler(); private final TestErrorHandler errorHandler = new TestErrorHandler();
private PollableChannel channelMock;
private final ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); private final ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
private PollingConsumer endpoint;
private PollableChannel channelMock;
@Before @Before
public void init() throws Exception { public void init() {
channelMock = Mockito.mock(PollableChannel.class); this.channelMock = mock(PollableChannel.class);
consumer.counter.set(0); this.endpoint = new PollingConsumer(this.channelMock, this.consumer);
trigger.reset(); this.taskScheduler.setPoolSize(5);
endpoint = new PollingConsumer(channelMock, consumer); this.endpoint.setErrorHandler(this.errorHandler);
taskScheduler.setPoolSize(5); this.endpoint.setTaskScheduler(this.taskScheduler);
endpoint.setErrorHandler(errorHandler); this.endpoint.setTrigger(this.trigger);
endpoint.setTaskScheduler(taskScheduler); this.endpoint.setBeanFactory(mock(BeanFactory.class));
endpoint.setTrigger(trigger); this.endpoint.setReceiveTimeout(-1);
endpoint.setBeanFactory(mock(BeanFactory.class)); this.endpoint.afterPropertiesSet();
endpoint.setReceiveTimeout(-1); this.taskScheduler.afterPropertiesSet();
endpoint.afterPropertiesSet();
taskScheduler.afterPropertiesSet();
} }
@After @After
public void stop() throws Exception { public void stop() {
taskScheduler.destroy(); taskScheduler.destroy();
} }
@Test @Test
public void singleMessage() { public void singleMessage() {
Mockito.when(channelMock.receive()).thenReturn(message); Mockito.when(this.channelMock.receive()).thenReturn(this.message);
endpoint.setMaxMessagesPerPoll(1); this.endpoint.setMaxMessagesPerPoll(1);
endpoint.start(); this.endpoint.start();
trigger.await(); this.trigger.await();
endpoint.stop(); this.endpoint.stop();
assertEquals(1, consumer.counter.get()); assertEquals(1, this.consumer.counter.get());
} }
@Test @Test
public void multipleMessages() { public void multipleMessages() {
Mockito.when(channelMock.receive()).thenReturn(message, message, message, message, message); Mockito.when(this.channelMock.receive())
endpoint.setMaxMessagesPerPoll(5); .thenReturn(this.message, this.message, this.message, this.message, this.message);
endpoint.start(); this.endpoint.setMaxMessagesPerPoll(5);
trigger.await(); this.endpoint.start();
endpoint.stop(); this.trigger.await();
assertEquals(5, consumer.counter.get()); this.endpoint.stop();
assertEquals(5, this.consumer.counter.get());
} }
@Test @Test
public void multipleMessages_underrun() { public void multipleMessages_under_run() {
Mockito.when(channelMock.receive()).thenReturn(message, message, message, message, message, null); Mockito.when(this.channelMock.receive())
endpoint.setMaxMessagesPerPoll(6); .thenReturn(this.message, this.message, this.message, this.message, this.message, null);
endpoint.start(); this.endpoint.setMaxMessagesPerPoll(6);
trigger.await(); this.endpoint.start();
endpoint.stop(); this.trigger.await();
assertEquals(5, consumer.counter.get()); this.endpoint.stop();
assertEquals(5, this.consumer.counter.get());
} }
@Test @Test
public void heavierLoadTest() throws Exception { public void heavierLoadTest() {
for (int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
this.init(); init();
this.multipleMessages(); this.trigger.reset();
this.stop(); this.consumer.counter.set(0);
multipleMessages();
stop();
} }
} }
@Test(expected = MessageRejectedException.class) @Test(expected = MessageRejectedException.class)
public void rejectedMessage() throws Throwable { public void rejectedMessage() throws Throwable {
Mockito.when(channelMock.receive()).thenReturn(badMessage); Mockito.when(this.channelMock.receive()).thenReturn(this.badMessage);
endpoint.start(); this.endpoint.start();
trigger.await(); this.trigger.await();
endpoint.stop(); this.endpoint.stop();
assertEquals(1, consumer.counter.get()); assertEquals(1, this.consumer.counter.get());
errorHandler.throwLastErrorIfAvailable(); this.errorHandler.throwLastErrorIfAvailable();
} }
@Test(expected = MessageRejectedException.class) @Test(expected = MessageRejectedException.class)
public void droppedMessage_onePerPoll() throws Throwable { public void droppedMessage_onePerPoll() throws Throwable {
Mockito.when(channelMock.receive()).thenReturn(badMessage); Mockito.when(this.channelMock.receive()).thenReturn(this.badMessage);
endpoint.setMaxMessagesPerPoll(10); this.endpoint.setMaxMessagesPerPoll(10);
endpoint.start(); this.endpoint.start();
trigger.await(); this.trigger.await();
endpoint.stop(); this.endpoint.stop();
assertEquals(1, consumer.counter.get()); assertEquals(1, this.consumer.counter.get());
errorHandler.throwLastErrorIfAvailable(); this.errorHandler.throwLastErrorIfAvailable();
} }
@Test @Test
public void blockingSourceTimedOut() { public void blockingSourceTimedOut() {
// we don't need to await the timeout, returning null suffices // we don't need to await the timeout, returning null suffices
Mockito.when(channelMock.receive()).thenReturn(null); Mockito.when(this.channelMock.receive()).thenReturn(null);
endpoint.setReceiveTimeout(1); this.endpoint.setReceiveTimeout(1);
endpoint.start(); this.endpoint.start();
trigger.await(); this.trigger.await();
endpoint.stop(); this.endpoint.stop();
assertEquals(0, consumer.counter.get()); assertEquals(0, this.consumer.counter.get());
} }
@Test @Test
public void blockingSourceNotTimedOut() { public void blockingSourceNotTimedOut() {
Mockito.when(channelMock.receive(Mockito.eq(1L))).thenReturn(message); Mockito.when(this.channelMock.receive(Mockito.eq(1L))).thenReturn(this.message);
endpoint.setReceiveTimeout(1); this.endpoint.setReceiveTimeout(1);
endpoint.setMaxMessagesPerPoll(1); this.endpoint.setMaxMessagesPerPoll(1);
endpoint.start(); this.endpoint.start();
trigger.await(); this.trigger.await();
endpoint.stop(); this.endpoint.stop();
assertEquals(1, consumer.counter.get()); assertEquals(1, this.consumer.counter.get());
} }
@@ -192,45 +188,6 @@ public class PollingConsumerEndpointTests {
} }
private static class TestTrigger implements Trigger {
private final AtomicBoolean hasRun = new AtomicBoolean();
private volatile CountDownLatch latch = new CountDownLatch(1);
TestTrigger() {
super();
}
@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
if (!this.hasRun.getAndSet(true)) {
return new Date();
}
this.latch.countDown();
return null;
}
public void reset() {
this.latch = new CountDownLatch(1);
this.hasRun.set(false);
}
public void await() {
try {
this.latch.await(5000, TimeUnit.MILLISECONDS);
if (latch.getCount() != 0) {
throw new RuntimeException("test latch.await() did not count down");
}
}
catch (InterruptedException e) {
throw new RuntimeException("test latch.await() interrupted");
}
}
}
private static class TestErrorHandler implements ErrorHandler { private static class TestErrorHandler implements ErrorHandler {
private volatile Throwable lastError; private volatile Throwable lastError;
@@ -252,6 +209,7 @@ public class PollingConsumerEndpointTests {
this.lastError = null; this.lastError = null;
throw t; throw t;
} }
} }
} }

7
spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2016 the original author or authors. * Copyright 2002-2018 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -28,6 +28,8 @@ import org.springframework.scheduling.TriggerContext;
* *
* @author Gunnar Hillert * @author Gunnar Hillert
* @author Gary Russell * @author Gary Russell
* @author Artem Bilan
*
* @since 2.2 * @since 2.2
* *
*/ */
@@ -47,7 +49,6 @@ public class OnlyOnceTrigger implements Trigger {
@Override @Override
public Date nextExecutionTime(TriggerContext triggerContext) { public Date nextExecutionTime(TriggerContext triggerContext) {
if (this.hasRun.getAndSet(true)) { if (this.hasRun.getAndSet(true)) {
this.latch.countDown(); this.latch.countDown();
return null; return null;
@@ -95,7 +96,7 @@ public class OnlyOnceTrigger implements Trigger {
public void await() { public void await() {
try { try {
if (!this.latch.await(5000, TimeUnit.MILLISECONDS)) { if (!this.latch.await(10000, TimeUnit.MILLISECONDS)) {
throw new RuntimeException("test latch.await() did not count down"); throw new RuntimeException("test latch.await() did not count down");
} }
} }