GCS - Object is missing required member 'access_token'

[francisdb]

Alpakka 2.0.0-M2

We sporadically see this error in our logs, the auth token handling is not handling error responses correctly. In this case a retry should be done like in #2057 and #1931

alpakka/google-cloud-storage/src/main/scala/akka/stream/alpakka/googlecloud/storage/impl/GoogleTokenApi.scala

Lines 53 to 54 in f420069

	response <- http.singleRequest(HttpRequest(HttpMethods.POST, settings.url, entity = requestEntity))
	result <- Unmarshal(response.entity).to[OAuthResponse]

Also apply to the google-cloud-pub-sub module

alpakka/google-cloud-pub-sub/src/main/scala/akka/stream/alpakka/googlecloud/pubsub/impl/GoogleTokenApi.scala

Lines 54 to 55 in f420069

	response <- http.singleRequest(HttpRequest(HttpMethods.POST, googleTokenUrl, entity = requestEntity))
	result <- Unmarshal(response.entity).to[OAuthResponse]

Also the trace for this error does not show it is related to alpakka?

play.api.UnexpectedException: Unexpected exception[DeserializationException: Object is missing required member 'access_token']
    at play.api.http.HttpErrorHandlerExceptions$.throwableToUsefulException(HttpErrorHandler.scala:328)
    at play.api.http.DefaultHttpErrorHandler.onServerError(HttpErrorHandler.scala:251)
    at play.core.server.AkkaHttpServer$$anonfun$2.applyOrElse(AkkaHttpServer.scala:421)
    at play.core.server.AkkaHttpServer$$anonfun$2.applyOrElse(AkkaHttpServer.scala:417)
    at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:453)
    at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
    at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:92)
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
    at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:94)
    at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:92)
    at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:47)
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:47)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
    at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:177)

spray.json.DeserializationException: Object is missing required member 'access_token'
    at spray.json.package$.deserializationError(package.scala:23)
    at spray.json.ProductFormats.fromField(ProductFormats.scala:61)
    at spray.json.ProductFormats.fromField$(ProductFormats.scala:51)
    at spray.json.DefaultJsonProtocol$.fromField(DefaultJsonProtocol.scala:30)
    at spray.json.ProductFormatsInstances$$anon$3.read(ProductFormatsInstances.scala:81)
    at spray.json.ProductFormatsInstances$$anon$3.read(ProductFormatsInstances.scala:71)
    at akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport.$anonfun$sprayJsonUnmarshaller$1(SprayJsonSupport.scala:31)
    at akka.http.scaladsl.util.FastFuture$.$anonfun$map$1(FastFuture.scala:23)
    at akka.http.scaladsl.util.FastFuture$.strictTransform$1(FastFuture.scala:41)
    at akka.http.scaladsl.util.FastFuture$.$anonfun$transformWith$3(FastFuture.scala:51)
    at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:447)
    at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
    at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:92)
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
    at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:94)
    at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:92)
    at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:47)
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:47)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
    at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:177)

java.util.NoSuchElementException: key not found: access_token
    at scala.collection.MapOps.default(Map.scala:246)
    at scala.collection.MapOps.default$(Map.scala:245)
    at scala.collection.AbstractMap.default(Map.scala:376)
    at scala.collection.MapOps.apply(Map.scala:148)
    at scala.collection.MapOps.apply$(Map.scala:147)
    at scala.collection.AbstractMap.apply(Map.scala:376)
    at spray.json.ProductFormats.fromField(ProductFormats.scala:58)
    at spray.json.ProductFormats.fromField$(ProductFormats.scala:51)
    at spray.json.DefaultJsonProtocol$.fromField(DefaultJsonProtocol.scala:30)
    at spray.json.ProductFormatsInstances$$anon$3.read(ProductFormatsInstances.scala:81)
    at spray.json.ProductFormatsInstances$$anon$3.read(ProductFormatsInstances.scala:71)
    at akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport.$anonfun$sprayJsonUnmarshaller$1(SprayJsonSupport.scala:31)
    at akka.http.scaladsl.util.FastFuture$.$anonfun$map$1(FastFuture.scala:23)
    at akka.http.scaladsl.util.FastFuture$.strictTransform$1(FastFuture.scala:41)
    at akka.http.scaladsl.util.FastFuture$.$anonfun$transformWith$3(FastFuture.scala:51)
    at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:447)
    at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
    at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:92)
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
    at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:94)
    at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:92)
    at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:47)
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:47)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
    at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:177)

[seglo]

Thanks for reporting this and #2057. I created a PR that addresses #2057 and will enable retries for any known server error code (5xx).

#2082

We sporadically see this error in our logs, the auth token handling is not handling error responses correctly. In this case a retry should be done like in #2057 and #1931

Are you suggesting the implementation to get the access token needs to handle a error responses instead of just OAuthResponse?

[francisdb]

Yes, from the error we see that the request clearly failed to return a valid response, my guess is that it also returned 5xx

adding a response code check + same retry logic in both above files should reduce the error rate

[francisdb]

currently #2082 is not fixing this issue

[seglo]

I see. It seems reasonable to assume that the unmarshalling step is what's causing the exception, but I agree it's odd that it's not in the stack trace. Would you be open to troubleshooting this further, or working on a PR? We should find a way to reproduce it with a mock in our tests.

[francisdb]

Will see if I find the time during the coming days