Memory pressure can rise during outbox draining

### Before opening, please confirm:

- [X] I have [searched for duplicate or closed issues](https://github.com/aws-amplify/amplify-android/issues?q=is%3Aissue+) and [discussions](https://github.com/aws-amplify/amplify-android/discussions).

### Language and Async Model

Java

### Amplify Categories

DataStore

### Gradle script dependencies

```groovy
// Put output below this line

1.37.5
```

### Environment information

```
# Put output below this line

------------------------------------------------------------
Gradle 6.8.2
------------------------------------------------------------

Build time: 2021-02-05 12:53:00 UTC
Revision: b9bd4a5c6026ac52f690eaf2829ee26563cad426

Kotlin: 1.4.20
Groovy: 2.5.12
Ant: Apache Ant(TM) version 1.10.9 compiled on September 27 2020
JVM: 11.0.17 (Oracle Corporation 11.0.17+1)
OS: Linux 6.0.2-arch1-1 amd64

```

### Please include any relevant guides or documentation you're referencing

_No response_

### Describe the bug

drainMutationOutbox blocks the storageChanges Observer during periods of long transmission causing invisible model back pressure to build in the application. Records will be recorded to the model tables but not the persistent record table until the draining is complete.

What happens is all the models bottleneck between [observation of the local storage changes to enqueuing one](https://github.com/aws-amplify/amplify-android/blob/e3b222d73ebec419f9630869cad127dffc1b88b1/aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/StorageObserver.java#L60-L74) and [processing the entire mutation outbox](https://github.com/aws-amplify/amplify-android/blob/release_v1.37.5/aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/MutationProcessor.java#L93-L103) before giving up the scheduler.single() thread. In bad connectivity the application will get out of memory errors.

The fix is to make [mutationOubox's observer](https://github.com/aws-amplify/amplify-android/blob/release_v1.37.5/aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/MutationProcessor.java#L102) work off Schedulers.io() vs the current Schedulers.single().

### Reproduction steps (if applicable)

I've been able to replicate this observation by running offline to 1700 pending records and then bring the device online and letting it drain to ~100 pending records. It appears that the drainMutationOutbox processing is effectively blocking until the outbox fully drains. In extreme instances if you have ~1700 records pending and then being a drain, even if the device goes offline, because the queue is pending it will not enqueue more due to the observation on Schedulesrs.single() being a single thread that the MutationProcessor.startDrainingMutationOutbox and the StorageObserver.startObservingStorageChanges compete for. Because the localStorageAdapter is emitting into the StorageObserver after a saving a model, the enqueueing of the model to the mutationOutbox, ie writing it as a PersistentRecord, is held up sharing the single scheduler. Moving the observation of the mutationOutbox to Schedulers.io seems to balance out this blocking in extreme circumstances.

### Code Snippet

```java
// Put your code below this line.

```

### Log output

```
// Put your logs below this line

```

### amplifyconfiguration.json

_No response_

### GraphQL Schema

```graphql
// Put your schema below this line

```

### Additional information and screenshots

_No response_

Start with StorageObserver.java at the storageChanges observer and MutationProcessor.java at startDrainingMutationOutbox and the mutationOutbox observer. Reproduce the issue with a large pending outbox and poor connectivity, then verify that storage changes continue being enqueued while the outbox drains and that the backlog does not build until the application runs out of memory.

Written by the indexing model from the issue text.