AWS Lambda + Mutiny + Thread blocked #31768
Unanswered
cthiebault
asked this question in
Q&A
Replies: 2 comments 2 replies
-
|
/cc @cescoffier (mutiny), @evanchooly (kotlin), @geoand (kotlin), @jponge (mutiny), @matejvasek (amazon-lambda), @patriot1burke (amazon-lambda) |
Beta Was this translation helpful? Give feedback.
0 replies
-
|
It looks like the method is called on the event loop. That's a bit surprising as the method has a synchronous signature. Can you tell us more about your setup? Providing a simple reproducer would be awesome! |
Beta Was this translation helpful? Give feedback.
2 replies
-
|
I will try to extract some code with same logic but I can't copy/paste my code here as it's not opensource :-/ |
Beta Was this translation helpful? Give feedback.
-
|
Here is my code: package demo
import com.amazonaws.services.lambda.runtime.Context
import com.amazonaws.services.lambda.runtime.RequestHandler
import com.amazonaws.services.lambda.runtime.events.SQSEvent
import com.fasterxml.jackson.databind.ObjectMapper
import io.quarkus.logging.Log
import io.smallrye.mutiny.Uni
import io.smallrye.mutiny.infrastructure.Infrastructure.getDefaultWorkerPool
import io.smallrye.mutiny.unchecked.Unchecked
import io.vertx.ext.auth.authentication.UsernamePasswordCredentials
import io.vertx.mutiny.core.buffer.Buffer
import io.vertx.mutiny.ext.web.client.WebClient
import org.eclipse.microprofile.config.inject.ConfigProperty
import javax.inject.Inject
class IndexerHandler @Inject constructor(
private val objectMapper: ObjectMapper,
private val webClient: WebClient,
@ConfigProperty(name = "opensearch.url") private val url: String,
@ConfigProperty(name = "opensearch.username") private val username: String,
@ConfigProperty(name = "opensearch.password") private val password: String,
) : RequestHandler<SQSEvent, Boolean> {
data class ElasticsearchDocument(
val indexName: String,
val id: String,
)
override fun handleRequest(event: SQSEvent, context: Context?): Boolean =
mapToElasticDocuments(event)
.onItem().transformToUni { documents -> bulkIndex(documents) }
.replaceWith(true)
.onFailure().invoke { error: Throwable -> Log.error("Cannot index documents", error) }
.await().indefinitely() // lambda timeout will occur anyway
private fun mapToElasticDocuments(event: SQSEvent): Uni<List<ElasticsearchDocument>> =
Uni.createFrom().item {
event.records.map { ElasticsearchDocument(indexName = "demo", id = it.messageId) }
}.runSubscriptionOn(getDefaultWorkerPool())
private fun bulkIndex(documents: List<ElasticsearchDocument>): Uni<Void> =
if (documents.isEmpty()) {
Uni.createFrom().voidItem()
} else {
postBulk(bulkJsonBody(documents))
}
private fun bulkJsonBody(documents: List<ElasticsearchDocument>) =
documents
.map { document ->
listOf(
mapOf("index" to mapOf("_index" to document.indexName, "_id" to document.id)),
document
)
}
.flatten()
.joinToString(separator = System.lineSeparator(), postfix = System.lineSeparator()) {
objectMapper.writeValueAsString(it)
} + System.lineSeparator()
private fun postBulk(command: String): Uni<Void> =
webClient.postAbs(url + (if (url.endsWith("/")) "" else "/") + "_bulk")
.authentication(UsernamePasswordCredentials(username, password))
.putHeader("content-type", "application/json")
.sendBuffer(Buffer.buffer(command))
.onItem().transform(Unchecked.function { response ->
if (response.statusCode() != 200) {
throw RuntimeException("Cannot bulk index (status: ${response.statusCode()}). Bulk request: $command")
}
}).replaceWithVoid()
}Looking at the logs, the Thread blocked warning seems to occur after the lambda is finished: Or |
Beta Was this translation helpful? Give feedback.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
-
Hello,
I'm migrating my Quarkus Lambda (triggered by SQS events) to Quarkus + Mutiny Lambda.
My function is calling an external web service and then processing the response. I wait indefinitely as the lambda has already a timeout defined.
But since I'm using Mutiny (with Vertx WebClient), I get
Thread blockedexceptions...I'm new to reactive programming but if I understand well, the main thread is waiting for the Uni to be completed... so it's normal that this thread is blocked, right?
Thanks for helping me to better understand what is happening :-)
Beta Was this translation helpful? Give feedback.
All reactions