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

GH-8562: Fix streaming source for remote calls (#8564)

Browse Source 
* GH-8562: Fix streaming source for remote calls

Fixes https://github.com/spring-projects/spring-integration/issues/8562

The `AbstractRemoteFileStreamingMessageSource.doReceive()` takes files first from a `toBeReceived` queue.
When `AbstractRemoteFileStreamingMessageSource.remoteFileToMessage()` fails to fetch the file content
because of interim connection issue, we reset this file from a filter and rethrow an exception.
The next `receive()` call will just go ahead to the next entry in the `toBeReceived` queue, but the
file we have just failed for will be retried only on the next list call to the remove directory.
This essentially breaks a possible in-order target application logic.

* Introduce `AbstractRemoteFileStreamingMessageSource.strictOrder` option to clear the `toBeReceived` queue
 when we fail in the `remoteFileToMessage()`, so the next `receive()`
 call would re-fetch files from remote dir, because the filter has been reset for those files.
* Fix `AbstractFileInfo.toString()` to not perform remote calls when we just log this file.
For example, we reset the file for connection failure and log the message about it,
but it fails again because we request `size` of the file which may require a remote connection.

**Cherry-pick to `6.0.x` & `5.5.x`**

* * Revert `AbstractFileInfo` changes
* Override `toString()` in `SmbFileInfo` instead -
exactly the place where connection is used to obtain
file attributes like `size` or `lastModified`
This commit is contained in:
Artem Bilan
2023-02-28 16:34:19 -05:00
committed by GitHub
parent a09d6dcd2c
commit ce9f7f4321
3 changed files with 49 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -243,6 +243,10 @@ public class StreamingInboundTests {
.isThrownBy(streamer::receive);
assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(1);
assertThat(streamer.metadataMap).hasSize(0);
streamer.setStrictOrder(true);
assertThatExceptionOfType(UncheckedIOException.class)
.isThrownBy(streamer::receive);
assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(0);
}
public static class Streamer extends AbstractRemoteFileStreamingMessageSource<String> {