=3974 per Persist (serialize) actor refs with transport info
* The reason for the problem with NoSuchElementException in ClusterSharding was that actor references were not serialized with full address information. In certain fail over scenarios the references could not be resolved and therefore the ShardRegionTerminated did not match corresponding ShardRegionRegistered. * Wrap serialization with transport information from defaultAddress (cherry picked from commit 3e73ae5925cf1293a9a5d61e48919b1708e84df2)
This commit is contained in:
parent
167fd4e738
commit
e4b2af3783
3 changed files with 53 additions and 19 deletions
|
|
@ -1198,7 +1198,13 @@ class ShardCoordinator(handOffTimeout: FiniteDuration, rebalanceInterval: Finite
|
||||||
case ShardRegionProxyRegistered(proxy) ⇒
|
case ShardRegionProxyRegistered(proxy) ⇒
|
||||||
persistentState = persistentState.updated(evt)
|
persistentState = persistentState.updated(evt)
|
||||||
case ShardRegionTerminated(region) ⇒
|
case ShardRegionTerminated(region) ⇒
|
||||||
persistentState = persistentState.updated(evt)
|
if (persistentState.regions.contains(region))
|
||||||
|
persistentState = persistentState.updated(evt)
|
||||||
|
else {
|
||||||
|
log.debug("ShardRegionTerminated, but region {} was not registered. This inconsistency is due to that " +
|
||||||
|
" some stored ActorRef in Akka v2.3.0 and v2.3.1 did not contain full address information. It will be " +
|
||||||
|
"removed by later watch.", region)
|
||||||
|
}
|
||||||
case ShardRegionProxyTerminated(proxy) ⇒
|
case ShardRegionProxyTerminated(proxy) ⇒
|
||||||
persistentState = persistentState.updated(evt)
|
persistentState = persistentState.updated(evt)
|
||||||
case _: ShardHomeAllocated ⇒
|
case _: ShardHomeAllocated ⇒
|
||||||
|
|
|
||||||
|
|
@ -36,6 +36,12 @@ class MessageSerializer(val system: ExtendedActorSystem) extends Serializer {
|
||||||
def identifier: Int = 7
|
def identifier: Int = 7
|
||||||
def includeManifest: Boolean = true
|
def includeManifest: Boolean = true
|
||||||
|
|
||||||
|
private lazy val transportInformation: Option[Serialization.Information] = {
|
||||||
|
val address = system.provider.getDefaultAddress
|
||||||
|
if (address.hasLocalScope) None
|
||||||
|
else Some(Serialization.Information(address, system))
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Serializes [[PersistentBatch]], [[PersistentRepr]] and [[Deliver]] messages. Delegates
|
* Serializes [[PersistentBatch]], [[PersistentRepr]] and [[Deliver]] messages. Delegates
|
||||||
* serialization of a persistent message's payload to a matching `akka.serialization.Serializer`.
|
* serialization of a persistent message's payload to a matching `akka.serialization.Serializer`.
|
||||||
|
|
@ -103,14 +109,22 @@ class MessageSerializer(val system: ExtendedActorSystem) extends Serializer {
|
||||||
}
|
}
|
||||||
|
|
||||||
private def persistentPayloadBuilder(payload: AnyRef) = {
|
private def persistentPayloadBuilder(payload: AnyRef) = {
|
||||||
val serializer = SerializationExtension(system).findSerializerFor(payload)
|
def payloadBuilder() = {
|
||||||
val builder = PersistentPayload.newBuilder()
|
val serializer = SerializationExtension(system).findSerializerFor(payload)
|
||||||
|
val builder = PersistentPayload.newBuilder()
|
||||||
|
|
||||||
if (serializer.includeManifest) builder.setPayloadManifest((ByteString.copyFromUtf8(payload.getClass.getName)))
|
if (serializer.includeManifest) builder.setPayloadManifest((ByteString.copyFromUtf8(payload.getClass.getName)))
|
||||||
|
|
||||||
builder.setPayload(ByteString.copyFrom(serializer.toBinary(payload)))
|
builder.setPayload(ByteString.copyFrom(serializer.toBinary(payload)))
|
||||||
builder.setSerializerId(serializer.identifier)
|
builder.setSerializerId(serializer.identifier)
|
||||||
builder
|
builder
|
||||||
|
}
|
||||||
|
|
||||||
|
// serialize actor references with full address information (defaultAddress)
|
||||||
|
transportInformation match {
|
||||||
|
case Some(ti) ⇒ Serialization.currentTransportInformation.withValue(ti) { payloadBuilder() }
|
||||||
|
case None ⇒ payloadBuilder()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private def deliveredMessageBuilder(delivered: Delivered) = {
|
private def deliveredMessageBuilder(delivered: Delivered) = {
|
||||||
|
|
|
||||||
|
|
@ -6,9 +6,9 @@
|
||||||
package akka.persistence.serialization
|
package akka.persistence.serialization
|
||||||
|
|
||||||
import java.io._
|
import java.io._
|
||||||
|
|
||||||
import akka.actor._
|
import akka.actor._
|
||||||
import akka.serialization.{ Serializer, SerializationExtension }
|
import akka.serialization.{ Serializer, SerializationExtension }
|
||||||
|
import akka.serialization.Serialization
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Wrapper for snapshot `data`. Snapshot `data` are the actual snapshot objects captured by
|
* Wrapper for snapshot `data`. Snapshot `data` are the actual snapshot objects captured by
|
||||||
|
|
@ -32,6 +32,12 @@ class SnapshotSerializer(system: ExtendedActorSystem) extends Serializer {
|
||||||
def identifier: Int = 8
|
def identifier: Int = 8
|
||||||
def includeManifest: Boolean = false
|
def includeManifest: Boolean = false
|
||||||
|
|
||||||
|
private lazy val transportInformation: Option[Serialization.Information] = {
|
||||||
|
val address = system.provider.getDefaultAddress
|
||||||
|
if (address.hasLocalScope) None
|
||||||
|
else Some(Serialization.Information(address, system))
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Serializes a [[Snapshot]]. Delegates serialization of snapshot `data` to a matching
|
* Serializes a [[Snapshot]]. Delegates serialization of snapshot `data` to a matching
|
||||||
* `akka.serialization.Serializer`.
|
* `akka.serialization.Serializer`.
|
||||||
|
|
@ -49,22 +55,30 @@ class SnapshotSerializer(system: ExtendedActorSystem) extends Serializer {
|
||||||
Snapshot(snapshotFromBinary(bytes))
|
Snapshot(snapshotFromBinary(bytes))
|
||||||
|
|
||||||
private def snapshotToBinary(snapshot: AnyRef): Array[Byte] = {
|
private def snapshotToBinary(snapshot: AnyRef): Array[Byte] = {
|
||||||
val extension = SerializationExtension(system)
|
def serialize() = {
|
||||||
|
val extension = SerializationExtension(system)
|
||||||
|
|
||||||
val snapshotSerializer = extension.findSerializerFor(snapshot)
|
val snapshotSerializer = extension.findSerializerFor(snapshot)
|
||||||
val snapshotManifest = if (snapshotSerializer.includeManifest) Some(snapshot.getClass.getName) else None
|
val snapshotManifest = if (snapshotSerializer.includeManifest) Some(snapshot.getClass.getName) else None
|
||||||
|
|
||||||
val header = SnapshotHeader(snapshotSerializer.identifier, snapshotManifest)
|
val header = SnapshotHeader(snapshotSerializer.identifier, snapshotManifest)
|
||||||
val headerSerializer = extension.findSerializerFor(header)
|
val headerSerializer = extension.findSerializerFor(header)
|
||||||
val headerBytes = headerSerializer.toBinary(header)
|
val headerBytes = headerSerializer.toBinary(header)
|
||||||
|
|
||||||
val out = new ByteArrayOutputStream
|
val out = new ByteArrayOutputStream
|
||||||
|
|
||||||
writeInt(out, headerBytes.length)
|
writeInt(out, headerBytes.length)
|
||||||
|
|
||||||
out.write(headerBytes)
|
out.write(headerBytes)
|
||||||
out.write(snapshotSerializer.toBinary(snapshot))
|
out.write(snapshotSerializer.toBinary(snapshot))
|
||||||
out.toByteArray
|
out.toByteArray
|
||||||
|
}
|
||||||
|
|
||||||
|
// serialize actor references with full address information (defaultAddress)
|
||||||
|
transportInformation match {
|
||||||
|
case Some(ti) ⇒ Serialization.currentTransportInformation.withValue(ti) { serialize() }
|
||||||
|
case None ⇒ serialize()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private def snapshotFromBinary(bytes: Array[Byte]): AnyRef = {
|
private def snapshotFromBinary(bytes: Array[Byte]): AnyRef = {
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue