Amplify-android: DataStore sync not downloading newly created data from DynamoDB

[upachecog]

Datastore and API version: 1.6.2, but I noticed this in previous versions.

Current behavior:

• When I execute mutations through AWS AppSync to create new items, those items aren't in the queries to the DataStore.
• Observe callbacks only get executed after I close and re-open the app.
• When I close and re-open the app, synchronization is executed again: (Shouldn't a query return the local data immediately after the first synchronization occurred, if I never called DataStore.clear() ?)

Current implementation:

I call this query several times to update my UI based on the latest data, but it doesn't return the newly created items.

 Amplify.DataStore.query(Location::class.java,
      Where.matches(Location.FECHA.eq(Utils.getCurrentDateString())),
      { matches ->
          Log.i(
             "TaskDataStore",
              "Obtained today's ${Utils.getCurrentDateString()} task list: $matches"
          )
          onResponse(matches.asSequence().toList())
      },
      { onFailure(it) }
)

I call this method in the onCreate method of my first Activity:

Amplify.DataStore.observe(Location::class.java,
      { Log.i("MyAmplifyApp", "Observation began.") },
      { Log.i("MyAmplifyApp", "New Location: ${it.item()}") },
      { Log.e("MyAmplifyApp", "Observation failed.", it) },
      { Log.i("MyAmplifyApp", "Observation complete.") }
)

When I open the app for the first time I can see the log: Observation began.
The log New Location: ${it.item()} never shows until I close and re-open the app.

Expected behavior:

• All newly created items are synced down to the local database.
• Observation callback is executed when new items are created and syncs down data from the remote DB.
• After the first sync, closing the app and re-opening it should return local data immediately.

Device:

Samsung Galaxy A8
Running Android 10

[richardmcclellan]

Hi @upachecog, this sounds like an error occurred internally which caused DataStore it to shutdown unexpectedly. Then, when you close and re-open the app again, it is re-initialized, and you are able to see the data.

Could you share any logcat output that contains logs from the Amplify library? That should help to narrow down the root cause.

[upachecog]

@richardmcclellan Thanks for your answer.

Here is a full log of the app, from the beginning until the first successful response of the DataStore.

It starts with an already signed-in user.

Sync.log

I really can't see any error or warning message that could give me a clue about what is going on. And I'm still facing this issue.

[jamesonwilliams]

Hi @upachecog, thanks for the logs. I just looked through it, and it looks overall like a healthy client session, to me. Can we get more details about how you're creating the new items? Are you doing so through the AWS AppSync _Console_? What mutations and input data do you supply, and what response do you get? Does the selection set you're using to do the mutation include all of the fields?

[upachecog]

Hi @jamesonwilliams

Are you doing so through the AWS AppSync Console?

Yes, I am using the AWS AppSync Console

What mutations and input data do you supply, and what response do you get?

This is a mutation example:

mutation crearUbicacion{
  createUbicacion(input: {
    estatus: PENDIENTE,
    fecha: "2020-11-24",
    sucursalId: "123456",
    nombre: "Some name",
    codigoPostal: "12345",
    ciudad: "Some city",
    entidadFederativa: "Some entity",
    municipio: "Some region",
    colonia: "Some town",
    calle: "Some street",
    numeroExterior: "1896",
    numeroInterior: "",
    entreCalles: "",
    pais: "Some country",
    coordenada: ["Coordinate1","Coordinate2"],
    asignado: "d32b4454-b512-4a1d-a1c6-35d38105ce0e",
    clasificacion: ["Strgin1", "String2", "Área", "G. String", "String", "III - String, -"]
    ruta: "100"
  }){
    id
  }
} 

This is a response example:

{
  "data": {
    "createUbicacion": {
      "id": "d32b4454-b512-4a1d-a1c6-35d38105ce0e"
    }
  }
}

Does the selection set you're using to do the mutation include all of the fields?

Yes, but some of them are empty

[jamesonwilliams]

@upachecog Right now, it looks like your mutation's selection set only contains id. Try including _all_ of the fields in the model. The AppSync Realtime service builds the subscription response from the data in the mutation response. So, you have to include all of the same fields in the selection set.

[upachecog]

@jamesonwilliams you're right, the selection response had only the id

I tried again, now including all the fields in mutations selection, and now it seems to be triggering something, but also it is causing the app to crash, with this error message:

E/AndroidRuntime: FATAL EXCEPTION: RxCachedThreadScheduler-26
    Process: com.invernaderolabs.checkin, PID: 10974
    io.reactivex.rxjava3.exceptions.UndeliverableException: The exception could not be delivered to the consumer because it has already canceled/disposed the flow or the exception has nowhere to go to begin with. Further reading: https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling | DataStoreException{message=Error during subscription., cause=ApiException{message=Subscription failed., cause=java.lang.NullPointerException: Attempt to invoke interface method 'java.lang.Object java.util.Map.get(java.lang.Object)' on a null object reference, recoverySuggestion=Sorry, we don't have a suggested fix for this error yet.}, recoverySuggestion=Evaluate details.}
        at io.reactivex.rxjava3.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
        at io.reactivex.rxjava3.internal.util.AtomicThrowable.tryAddThrowableOrReport(AtomicThrowable.java:52)
        at io.reactivex.rxjava3.internal.operators.observable.ObservableFlatMap$InnerObserver.onError(ObservableFlatMap.java:531)
        at io.reactivex.rxjava3.internal.observers.BasicFuseableObserver.onError(BasicFuseableObserver.java:100)
        at io.reactivex.rxjava3.internal.observers.BasicFuseableObserver.onError(BasicFuseableObserver.java:100)
        at io.reactivex.rxjava3.internal.observers.BasicFuseableObserver.onError(BasicFuseableObserver.java:100)
        at io.reactivex.rxjava3.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.checkTerminated(ObservableObserveOn.java:281)
        at io.reactivex.rxjava3.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:172)
        at io.reactivex.rxjava3.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:255)
        at io.reactivex.rxjava3.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:65)
        at io.reactivex.rxjava3.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:56)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:301)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)
        at java.lang.Thread.run(Thread.java:764)
     Caused by: DataStoreException{message=Error during subscription., cause=ApiException{message=Subscription failed., cause=java.lang.NullPointerException: Attempt to invoke interface method 'java.lang.Object java.util.Map.get(java.lang.Object)' on a null object reference, recoverySuggestion=Sorry, we don't have a suggested fix for this error yet.}, recoverySuggestion=Evaluate details.}
        at com.amplifyframework.datastore.appsync.AppSyncClient.lambda$subscription$3(AppSyncClient.java:298)
        at com.amplifyframework.datastore.appsync.-$$Lambda$AppSyncClient$797ziDK0io-qXODzROLOA77stS8.accept(Unknown Source:4)
        at com.amplifyframework.api.aws.SubscriptionOperation.lambda$null$1$SubscriptionOperation(SubscriptionOperation.java:94)
        at com.amplifyframework.api.aws.-$$Lambda$SubscriptionOperation$l9lC7Ew6niNmyszMXQSUKS6YhwE.accept(Unknown Source:4)
        at com.amplifyframework.api.aws.SubscriptionEndpoint$Subscription.dispatchError(SubscriptionEndpoint.java:396)
        at com.amplifyframework.api.aws.SubscriptionEndpoint.notifyError(SubscriptionEndpoint.java:211)
        at com.amplifyframework.api.aws.SubscriptionEndpoint.access$100(SubscriptionEndpoint.java:60)
        at com.amplifyframework.api.aws.SubscriptionEndpoint$AmplifyWebSocketListener.onFailure(SubscriptionEndpoint.java:492)
        at okhttp3.internal.ws.RealWebSocket.failWebSocket(RealWebSocket.kt:592)
        at okhttp3.internal.ws.RealWebSocket$connect$1.onResponse(RealWebSocket.kt:197)
        at okhttp3.internal.connection.RealCall$AsyncCall.run(RealCall.kt:519)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167) 
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641) 
        at java.lang.Thread.run(Thread.java:764) 
E/AndroidRuntime: Caused by: ApiException{message=Subscription failed., cause=java.lang.NullPointerException: Attempt to invoke interface method 'java.lang.Object java.util.Map.get(java.lang.Object)' on a null object reference, recoverySuggestion=Sorry, we don't have a suggested fix for this error yet.}
        ... 9 more
    Caused by: java.lang.NullPointerException: Attempt to invoke interface method 'java.lang.Object java.util.Map.get(java.lang.Object)' on a null object reference
        at com.amplifyframework.datastore.appsync.AppSyncExtensions.<init>(AppSyncExtensions.java:48)
        at com.amplifyframework.datastore.syncengine.SubscriptionProcessor.isUnauthorizedException(SubscriptionProcessor.java:160)
        at com.amplifyframework.datastore.syncengine.SubscriptionProcessor.lambda$null$2$SubscriptionProcessor(SubscriptionProcessor.java:188)
        at com.amplifyframework.datastore.syncengine.-$$Lambda$SubscriptionProcessor$M1sNKll70iq_9O01UPeTz7ejWc0.accept(Unknown Source:13)
        at com.amplifyframework.datastore.appsync.AppSyncClient.lambda$subscription$2(AppSyncClient.java:289)
        at com.amplifyframework.datastore.appsync.-$$Lambda$AppSyncClient$8oi7Ep-KMdlaKCEkwdqbq2VCcd8.accept(Unknown Source:8)
        at com.amplifyframework.api.aws.SubscriptionEndpoint$Subscription.dispatchNextMessage(SubscriptionEndpoint.java:389)
        at com.amplifyframework.api.aws.SubscriptionEndpoint.notifySubscriptionData(SubscriptionEndpoint.java:227)
        at com.amplifyframework.api.aws.SubscriptionEndpoint.access$700(SubscriptionEndpoint.java:60)
        at com.amplifyframework.api.aws.SubscriptionEndpoint$AmplifyWebSocketListener.processJsonMessage(SubscriptionEndpoint.java:567)
        at com.amplifyframework.api.aws.SubscriptionEndpoint$AmplifyWebSocketListener.onMessage(SubscriptionEndpoint.java:473)
        at okhttp3.internal.ws.RealWebSocket.onReadMessage(RealWebSocket.kt:333)
        at okhttp3.internal.ws.WebSocketReader.readMessageFrame(WebSocketReader.kt:245)
        at okhttp3.internal.ws.WebSocketReader.processNextFrame(WebSocketReader.kt:106)
        at okhttp3.internal.ws.RealWebSocket.loopReader(RealWebSocket.kt:293)
        at okhttp3.internal.ws.RealWebSocket$connect$1.onResponse(RealWebSocket.kt:195)
            ... 4 more

I'll take a look to see if there's an issue already opened about this.

[jamesonwilliams]

Hm, yes, does look like a bug, though. Normal usage of the library shouldn't cause this:

Caused by: java.lang.NullPointerException: Attempt to invoke interface method 'java.lang.Object java.util.Map.get(java.lang.Object)' on a null object reference

[upachecog]

Do you think is something wrong with our models or configuration?

Maybe this Sentry report can help somehow.

[jamesonwilliams]

Ah yea, I just ran into this while doing some unrelated work.

I believe this is a regression, cause by #942, which went out in Amplify Android 1.5.0.

Workaround for right now is to stay on 1.4.2, the last release before the regression.

I will put up a PR to fix, momentarily.

[upachecog]

@jamesonwilliams

Thanks to your PR I was able to know what was the error with my mutation (one filed was missing), but the sync is still not working. I need to close and reopen the app to see the created data.

Could you help me with that, please?

I think I may be missing something.

[jamesonwilliams]

Hi @upachecog, did you move to 1.6.5? What do you see in the logs during app startup and sync? Any errors?

[upachecog]

Hi @jamesonwilliams thanks for your quick response.

Yes I moved to 1.6.5 and the crash went away, and the logs seem to be the same without any visible errors.

I just execute the mutation in the AWS AppSync Console and nothing happens in the app, and I am executing it every minute.

[upachecog]

Hey @jamesonwilliams, I found this error log when calling the observe.

Do you think it could be somehow related?

E/MyAmplifyApp: Observation failed.
    DataStoreException{message=Unable to acquire orchestrator lock. Transition currently in progress., cause=null, recoverySuggestion=Retry your request}
        at com.amplifyframework.datastore.syncengine.Orchestrator.start(Orchestrator.java:195)
        at com.amplifyframework.datastore.AWSDataStorePlugin.lambda$start$9$AWSDataStorePlugin(AWSDataStorePlugin.java:259)
        at com.amplifyframework.datastore.-$$Lambda$AWSDataStorePlugin$FPjeqf-ENbST6bEgtzMenBBWB54.call(Unknown Source:6)
        at com.amplifyframework.datastore.-$$Lambda$TpN5VBwI7X4qLhZPz2Xh_RciFYY.run(Unknown Source:2)
        at io.reactivex.rxjava3.internal.observers.CallbackCompletableObserver.onComplete(CallbackCompletableObserver.java:53)
        at io.reactivex.rxjava3.internal.operators.completable.CompletableSubscribeOn$SubscribeOnObserver.onComplete(CompletableSubscribeOn.java:79)
        at io.reactivex.rxjava3.internal.operators.completable.CompletableTimeout$TimeOutObserver.onComplete(CompletableTimeout.java:87)
        at io.reactivex.rxjava3.internal.operators.completable.CompletableCreate$Emitter.onComplete(CompletableCreate.java:65)
        at com.amplifyframework.datastore.AWSDataStorePlugin.lambda$waitForInitialization$7$AWSDataStorePlugin(AWSDataStorePlugin.java:241)
        at com.amplifyframework.datastore.-$$Lambda$AWSDataStorePlugin$F7opBt57ZUb5PGbudjZ5clNE1H8.subscribe(Unknown Source:2)
        at io.reactivex.rxjava3.internal.operators.completable.CompletableCreate.subscribeActual(CompletableCreate.java:40)
        at io.reactivex.rxjava3.core.Completable.subscribe(Completable.java:2850)
        at io.reactivex.rxjava3.internal.operators.completable.CompletableTimeout.subscribeActual(CompletableTimeout.java:53)
        at io.reactivex.rxjava3.core.Completable.subscribe(Completable.java:2850)
        at io.reactivex.rxjava3.internal.operators.completable.CompletableSubscribeOn$SubscribeOnObserver.run(CompletableSubscribeOn.java:64)
        at io.reactivex.rxjava3.core.Scheduler$DisposeTask.run(Scheduler.java:614)
        at io.reactivex.rxjava3.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:65)
        at io.reactivex.rxjava3.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:56)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:301)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)
        at java.lang.Thread.run(Thread.java:764)

Please, let me know if you have some idea about how can I fix it.

[jamesonwilliams]

Transition currently in progress.

@upachecog I would wait for the READY event before calling observe().

Amplify.Hub.subscribe(HubChannel.DATASTORE,
    { event -> event.getName().equals(DataStoreChannelEventName.READY.toString()) },
    { onDataStoreReady() }
)

fun onDataStoreReady() {
    // This function can be called whatever you want
    // It gets called once when the datastore is done initializing, as above.
    Amplify.DataStore.observe(...)
}