How to perform a series of steps in a single thread, with an async flow in spring-integration?

I currently have a spring-integration (v4.3.24) flow that looks like the following:

           |
           | list of
           | filepaths
      +----v---+
      |splitter|
      +----+---+
           | filepath
           |
+----------v----------+
|sftp-outbound-gateway|
|        "get"        |
+----------+----------+
           | file
+---------------------+
|     +----v----+     |
|     |decryptor|     |
|     +----+----+     |
|          |          |
|    +-----v------+   | set of transformers
|    |decompressor|   | (with routers before them
|    +-----+------+   | because some steps are optional)
|          |          | that process the file;
|       +--v--+       | call this "FileProcessor"
|       | ... |       |
|       +--+--+       |
+---------------------+
           |
      +----v----+
      |save file|
      | to disk |
      +----+----+
           |

All of the channels above are DirectChannels - Yup, I know this is a poor structure. This was working fine for files in small numbers. But now, I have to deal with thousands of files which need to go through the same flow - benchmarks reveal that this takes ~ 1 day to finish processing. So, I'm planning to introduce some parallel processing to this flow. I want to modify my flow to achieve something like this:

                                    |
                                    |
                         +----------v----------+
                         |sftp-outbound-gateway|
                         |       "mget"        |
                         +----------+----------+
                                    | list of files
                                    |
                               +----v---+
                               |splitter|
                               +----+---+
         one thread             one | thread        ...
           +------------------------+---------------+--+--+--+--+
           | file                   | file          |  |  |  |  |
+---------------------+  +---------------------+
|     +----v----+     |  |     +----v----+     |
|     |decryptor|     |  |     |decryptor|     |
|     +----+----+     |  |     +----+----+     |
|          |          |  |          |          |
|    +-----v------+   |  |    +-----v------+   |   ...
|    |decompressor|   |  |    |decompressor|   |
|    +-----+------+   |  |    +-----+------+   |
|          |          |  |          |          |
|       +--v--+       |  |       +--v--+       |
|       | ... |       |  |       | ... |       |
|       +--+--+       |  |       +--+--+       |
+---------------------+  +---------------------+
           |                        |
      +----v----+              +----v----+
      |save file|              |save file|
      | to disk |              | to disk |
      +----+----+              +----+----+
           |                        |
           |                        |

For parallel processing, I output the files from the splitter on to a ExecutorChannel with a ThreadPoolTaskExecutor.

Some of the questions that I have:

1. I want all of the "FileProcessor" steps for one file to happen on the same thread, while multiple files are processed in parallel. How can I achieve this?
   I saw from this answer, that a ExecutorChannel to MessageHandlerChain flow would offer such functionality. But, some of the steps inside "FileProcessor" are optional (using selector-expression with routers to skip some of the steps) - ruling out using a MessageHandlerChain. I can rig up a couple of MessageHandlerChains with Filters inside, but this more or less becomes the approach mentioned in #2.

2. If #1 cannot be achieved, will changing all of the channel types starting from the splitter, from DirectChannel to ExecutorChannel help in introducing some parallelism? If yes, should I create a new TaskExecutor for each channel or can I reuse one TaskExecutor bean for all channels (I cannot set scope="prototype" on a TaskExecutor bean)?

3. In your opinion, which approach (#1 or #2) is better? Why?

4. If I perform global error handling, like the approach mentioned here, will the other files continue to process even if one file errors out?