How does Lagom get entityId? is it thread safe?

Lagom
1

Hello all,

This may sound a bit crazy but I am starting to suspect that the way Lagom retrieves the entityId of the given PersistentEntity may not be thread-safe.

Let’s take a look at this event as an example:

        {
            "eventName": "SomeObjectEvents$SomeObjectCreatedEvt",
            "persistentId": "SomeObjectEntity|ce01baef-9a89-300e-bc5b-2c531db81fe4",
            "partitionNumber": 0,
            "sequenceNumber": 1,
            "timeStamp": "2020-01-14T02:51:05.696Z",
            "offSet": "b57b6600-3678-11ea-8d98-db76639dbe44",
            "event": {
                "SomeObject": {
                    "id": "c8e88ced-91dc-382f-a844-719720702e56",
                    "status": "ACTIVE",
                    "createdAt": "2020-01-14T10:51:05.694+08:00[Asia/Singapore]",
                    "email": "xxxxxxxx@gmail.com"
                }
            },
            "serializationManifest": "com.zzz.xxx.SomeObject.impl.SomeObject.ces.SomeObjectEvents$SomeObjectCreatedEvt"
        }

You will notice that persistence id is “SomeObjectEntity|ce01baef-9a89-300e-bc5b-2c531db81fe4”, and if you look at the SomeObject inside the event, it has the id “c8e88ced-91dc-382f-a844-719720702e56”.

What I am confused here is that the way this persistent entity decides on the “id” is by calling the entityId inside the actor is relying on calling the entityId variable inside the Persistent Entity like below.

It generates the id by:

val id = UUID.fromString(entityId)

And I checked it several times that this is the ONLY place where the id of this SomeObjectEntity gets decided.

I might be barking at the wrong tree here and I am still quite skeptical about my hypothesis but if you can shed some light on how Lagom retrieves the entityId and maybe if you can take a minute to think about the possibility of the thread-safety breach in the process, it would be really appreciated.

Thanks,

1 Like
2

Hi Joo,

I’ve just taken a look at the code myself, and I’d be surprised if there was a bug in Lagom here, the entity ID, according to Lagom, can’t change. However, I think there is an opportunity for misconfiguration that could be impacting you and causing this. You’re using Scala right? When you register a persistent entity, you pass in a factory for creating that persistent entity:

If that factory doesn’t create a new instance every time, but instead returns shared instances, then you will see exactly the problem you are describing. The entity id gets injecting into your entity using this internalSetEntityId method:

So, your entity is instantiated here when the actor is constructed:

And then in the actors constructor, you can see the method to set the entity being initialised and called here:

The same entity id is what’s returned from the overridden persistenceId method, and this is what Akka persistence uses to store as the events persistence id in the database, so there’s no opportunity for them to differ, unless something calls internalSetEntityId later, and the only way I can see that happening is if the entity factory returns the same entity instance twice, rather than constructing a new one each time.

3

Thanks a lot for your help James. Much appreciated it!

I think it would be helpful to provide further context so the community can learn from our mistakes and avoid this kind of misconfigurations.

We have been using our own custom wrapper called “EntityRefProvider” just for simplicity.

So we have this class

class EntityRefProvider(registry: PersistentEntityRegistry)(implicit ec: ExecutionContext) {

  def customerEntityRef(customerId: UUID): EntityRefWrapper[CustomerCmd] = {
    EntityRefWrapper(registry.refFor[CustomerEntity](customerId.toString))
  }
}

case class EntityRefWrapper[Command](ref: PersistentEntityRef[Command]) {
  def ask[Cmd <: Command with PersistentEntity.ReplyType[_]](command: Cmd): Future[command.ReplyType] = {
    ref.ask(command)
  }
}

Which gets “wired” in the CustomerComponents like this:

lazy val persistentEntityRefProvider: EntityRefProvider = wire[EntityRefProvider]

And then we use it send a command to the actor like this:

refProvider.customerEntityRef(request.persistenceEntityId).ask(SomeInterestingCommand(xyz))

4

The entity ref provider looks fine, I’m more interested the place where PersistentEntityRegistry.register is invoked, which is usually somewhere in your application loader.

5

Yes, it is invoked within the application loader

abstract class CustomerApplication(context: LagomApplicationContext)
    extends LagomApplication(context)
    with CustomerComponents
    with AhcWSComponents
    with LagomKafkaComponents {

  lazy val transactionService: TransactionService = serviceClient.implement[TransactionService]
  lazy val communicationService: CommunicationService = serviceClient.implement[CommunicationService]
  lazy val customerEntity: CustomerEntity = wire[CustomerEntity]

  persistentEntityRegistry.register(customerEntity)
}
6

Yep, so there’s the problem. There is only one instance of customerEntity that you are sharing between all persistent entities, stored in the lazy val, so when that instance gets started in one entity, and then another entity is started, the same instance is used so the new entity key is injected into that existing instance. Either change it to:

def customerEntity: CustomerEntity = wire[CustomerEntity]

Or probably better:

persistentEntityRegistry.register(wire[CustomerEntity])

And that will fix it.

7

Ah… precisely… we were able to replicate that issue in our environment.
Thanks a lot for your help James. It was really helpful.
We decided to go with “def” and name it def xyzEntityFactory:XyzEntity = wire[XyzEntity] just to be more expressive.

Thanks!!

8

Great. I’ve raised this issue to address the fact that this mistake is so easy to make:

1 Like