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

Improve RotatingServersTests for race conditions

Browse Source 
Related to https://build.spring.io/browse/INT-MASTERSPRING40-1314/

When we initialize a `CachingSessionFactory`, we may fail connecting
ot FTP server, but we realize that only when we call `super.getSession()`
from spied factory beans in the test configuration

* Move `super.getSession()` as a first call of the overridden method
and don't change the state of monitored variables when an exception
happens during FTP connection
This commit is contained in:
Artem Bilan
2021-01-07 10:56:48 -05:00
parent e77d3393a4
commit d4bf66f582
1 changed files with 80 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java
View File

@@ -70,7 +70,7 @@ import org.springframework.messaging.Message;
*/
public class RotatingServersTests extends FtpTestSupport {
private static String tmpDir = getLocalTempFolder().getAbsolutePath() + File.separator + "multiSF";
private static final String TMP_DIR = getLocalTempFolder().getAbsolutePath() + File.separator + "multiSF";
@BeforeEach
public void setup() {
@@ -111,7 +111,7 @@ public class RotatingServersTests extends FtpTestSupport {
@BeforeEach
@AfterEach
public void clean(TestInfo info) {
recursiveDelete(new File(tmpDir), info);
recursiveDelete(new File(TMP_DIR), info);
}
@AfterEach
@@ -132,85 +132,88 @@ public class RotatingServersTests extends FtpTestSupport {
@Test
public void testStandard() throws Exception {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(StandardConfig.class);
StandardConfig config = ctx.getBean(StandardConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
List<Integer> sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList());
assertThat(sfCalls).containsExactly(1, 1, 1, 2, 2, 2, 3, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1);
File f1 = new File(tmpDir + File.separator + "standard" + File.separator + "f1");
assertThat(f1.exists()).isTrue();
File f2 = new File(tmpDir + File.separator + "standard" + File.separator + "f2");
assertThat(f2.exists()).isTrue();
File f3 = new File(tmpDir + File.separator + "standard" + File.separator + "f3");
assertThat(f3.exists()).isTrue();
assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3);
ctx.close();
try (AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(StandardConfig.class)) {
StandardConfig config = ctx.getBean(StandardConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
List<Integer> sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList());
assertThat(sfCalls).containsExactly(1, 1, 1, 2, 2, 2, 3, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1);
File f1 = new File(TMP_DIR + File.separator + "standard" + File.separator + "f1");
assertThat(f1.exists()).isTrue();
File f2 = new File(TMP_DIR + File.separator + "standard" + File.separator + "f2");
assertThat(f2.exists()).isTrue();
File f3 = new File(TMP_DIR + File.separator + "standard" + File.separator + "f3");
assertThat(f3.exists()).isTrue();
assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3);
}
}
@Test
public void testFair() throws Exception {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(FairConfig.class);
StandardConfig config = ctx.getBean(StandardConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
List<Integer> sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList());
assertThat(sfCalls).containsExactly(1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3);
File f1 = new File(tmpDir + File.separator + "fair" + File.separator + "f1");
assertThat(f1.exists()).isTrue();
File f2 = new File(tmpDir + File.separator + "fair" + File.separator + "f2");
assertThat(f2.exists()).isTrue();
File f3 = new File(tmpDir + File.separator + "fair" + File.separator + "f3");
assertThat(f3.exists()).isTrue();
assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3);
ctx.close();
try (AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(FairConfig.class)) {
StandardConfig config = ctx.getBean(StandardConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
List<Integer> sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList());
assertThat(sfCalls).containsExactly(1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3);
File f1 = new File(TMP_DIR + File.separator + "fair" + File.separator + "f1");
assertThat(f1.exists()).isTrue();
File f2 = new File(TMP_DIR + File.separator + "fair" + File.separator + "f2");
assertThat(f2.exists()).isTrue();
File f3 = new File(TMP_DIR + File.separator + "fair" + File.separator + "f3");
assertThat(f3.exists()).isTrue();
assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3);
}
}
@Test
public void testVariableLocalDir() throws Exception {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(VariableLocalConfig.class);
StandardConfig config = ctx.getBean(StandardConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
List<Integer> sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList());
assertThat(sfCalls).containsExactly(1, 1, 1, 2, 2, 2, 3, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1);
File f1 = new File(tmpDir + File.separator + "variable" + File.separator + "foo" + File.separator + "f1");
assertThat(f1.exists()).isTrue();
File f2 = new File(tmpDir + File.separator + "variable" + File.separator + "baz" + File.separator + "f2");
assertThat(f2.exists()).isTrue();
File f3 = new File(tmpDir + File.separator + "variable" + File.separator + "fiz" + File.separator + "f3");
assertThat(f3.exists()).isTrue();
assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3);
ctx.close();
try (AnnotationConfigApplicationContext ctx =
new AnnotationConfigApplicationContext(VariableLocalConfig.class)) {
StandardConfig config = ctx.getBean(StandardConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
List<Integer> sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList());
assertThat(sfCalls).containsExactly(1, 1, 1, 2, 2, 2, 3, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1);
File f1 = new File(TMP_DIR + File.separator + "variable" + File.separator + "foo" + File.separator + "f1");
assertThat(f1.exists()).isTrue();
File f2 = new File(TMP_DIR + File.separator + "variable" + File.separator + "baz" + File.separator + "f2");
assertThat(f2.exists()).isTrue();
File f3 = new File(TMP_DIR + File.separator + "variable" + File.separator + "fiz" + File.separator + "f3");
assertThat(f3.exists()).isTrue();
assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3);
}
}
@Test
public void testStreaming() throws Exception {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(StreamingConfig.class);
StandardConfig config = ctx.getBean(StandardConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
List<Integer> sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList());
// there's an extra getSession() with this adapter in listFiles
assertThat(sfCalls).containsExactly(1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 1, 1, 2, 2, 3);
QueueChannel files = ctx.getBean("files", QueueChannel.class);
assertThat(files.getQueueSize()).isEqualTo(3);
Message<?> received = files.receive(0);
StaticMessageHeaderAccessor.getCloseableResource(received).close();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f1");
received = files.receive(0);
StaticMessageHeaderAccessor.getCloseableResource(received).close();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f2");
received = files.receive(0);
StaticMessageHeaderAccessor.getCloseableResource(received).close();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f3");
ctx.close();
try (AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(StreamingConfig.class)) {
StandardConfig config = ctx.getBean(StreamingConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
List<Integer> sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList());
// there's an extra getSession() with this adapter in listFiles
assertThat(sfCalls).containsExactly(1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 1, 1, 2, 2, 3);
QueueChannel files = ctx.getBean("files", QueueChannel.class);
assertThat(files.getQueueSize()).isEqualTo(3);
Message<?> received = files.receive(0);
StaticMessageHeaderAccessor.getCloseableResource(received).close();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f1");
received = files.receive(0);
StaticMessageHeaderAccessor.getCloseableResource(received).close();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f2");
received = files.receive(0);
StaticMessageHeaderAccessor.getCloseableResource(received).close();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f3");
}
}
@Test
public void testFairStreaming() throws Exception {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(FairStreamingConfig.class);
try {
try (AnnotationConfigApplicationContext ctx =
new AnnotationConfigApplicationContext(FairStreamingConfig.class)) {
StandardConfig config = ctx.getBean(StandardConfig.class);
assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
ctx.getBean(SourcePollingChannelAdapter.class).stop();
@@ -237,9 +240,6 @@ public class RotatingServersTests extends FtpTestSupport {
StaticMessageHeaderAccessor.getCloseableResource(received).close();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f6");
}
finally {
ctx.close();
}
}
@Configuration
@@ -252,13 +252,14 @@ public class RotatingServersTests extends FtpTestSupport {
@Bean
public SessionFactory<FTPFile> factory1() {
return new CachingSessionFactory<FTPFile>(rawSessionFactory()) {
return new CachingSessionFactory<>(rawSessionFactory()) {
@Override
public Session<FTPFile> getSession() {
Session<FTPFile> session = super.getSession();
StandardConfig.this.sessionSources.add(1);
StandardConfig.this.latch.countDown();
return super.getSession();
return session;
}
};
@@ -266,13 +267,14 @@ public class RotatingServersTests extends FtpTestSupport {
@Bean
public SessionFactory<FTPFile> factory2() {
return new CachingSessionFactory<FTPFile>(rawSessionFactory()) {
return new CachingSessionFactory<>(rawSessionFactory()) {
@Override
public Session<FTPFile> getSession() {
Session<FTPFile> session = super.getSession();
StandardConfig.this.sessionSources.add(2);
StandardConfig.this.latch.countDown();
return super.getSession();
return session;
}
};
@@ -280,13 +282,14 @@ public class RotatingServersTests extends FtpTestSupport {
@Bean
public SessionFactory<FTPFile> factory3() {
return new CachingSessionFactory<FTPFile>(rawSessionFactory()) {
return new CachingSessionFactory<>(rawSessionFactory()) {
@Override
public Session<FTPFile> getSession() {
Session<FTPFile> session = super.getSession();
StandardConfig.this.sessionSources.add(3);
StandardConfig.this.latch.countDown();
return super.getSession();
return session;
}
};
@@ -334,7 +337,7 @@ public class RotatingServersTests extends FtpTestSupport {
}
protected File localDir() {
return new File(tmpDir, "standard");
return new File(TMP_DIR, "standard");
}
}
@@ -349,7 +352,7 @@ public class RotatingServersTests extends FtpTestSupport {
@Override
protected File localDir() {
return new File(tmpDir, "fair");
return new File(TMP_DIR, "fair");
}
}
@@ -362,7 +365,7 @@ public class RotatingServersTests extends FtpTestSupport {
public IntegrationFlow flow() {
return IntegrationFlows.from(Ftp.inboundAdapter(sf())
.filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate"))
.localDirectory(new File(tmpDir, "variable"))
.localDirectory(new File(TMP_DIR, "variable"))
.localFilenameExpression("#remoteDirectory + T(java.io.File).separator + #root")
.remoteDirectory("."),
e -> e.poller(Pollers.fixedDelay(1).advice(advice())))