From d4bf66f582dcfb5a6cd8b20ac973a9c85d12ac92 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 7 Jan 2021 10:56:48 -0500 Subject: [PATCH] Improve RotatingServersTests for race conditions 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 --- .../ftp/inbound/RotatingServersTests.java | 157 +++++++++--------- 1 file changed, 80 insertions(+), 77 deletions(-) diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java index 5698571e28..e8d697a98c 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java @@ -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 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 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 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 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 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 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 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 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 factory1() { - return new CachingSessionFactory(rawSessionFactory()) { + return new CachingSessionFactory<>(rawSessionFactory()) { @Override public Session getSession() { + Session 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 factory2() { - return new CachingSessionFactory(rawSessionFactory()) { + return new CachingSessionFactory<>(rawSessionFactory()) { @Override public Session getSession() { + Session 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 factory3() { - return new CachingSessionFactory(rawSessionFactory()) { + return new CachingSessionFactory<>(rawSessionFactory()) { @Override public Session getSession() { + Session 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())))