raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

API changes related to DataDeleted, #27371

Browse source 
* The reason for this change is that `DataDeleted` didn't extend the
  `UpdateResponse` and `GetResponse` types and could therefore cause problems
  when `Update` and `Get` were used with `ask`. This was also a problem for
  Akka Typed.
* Introduce new messages types UpdateDataDeleted and GetDataDeleted
* Introduce SubscribeResponse because the responses can be both `Changed`
  and `Deleted` are responses to subscriptions. Important for Typed.
This commit is contained in:
Patrik Nordwall 2019-08-16 07:33:41 +02:00
parent 4576835cce
commit 1d09d2725b
11 changed files with 228 additions and 97 deletions
Download patch file Download diff file Expand all files Collapse all files

54
akka-distributed-data/src/main/scala/akka/cluster/ddata/Replicator.scala
View file

@ -540,6 +540,11 @@ object Replicator {
extends GetResponse[A]
with ReplicatorMessage
/**
* The [[Get]] request couldn't be performed because the entry has been deleted.
*/
final case class GetDataDeleted[A <: ReplicatedData](key: Key[A], request: Option[Any]) extends GetResponse[A]
/**
* Register a subscriber that will be notified with a [[Changed]] message
* when the value of the given `key` is changed. Current value is also
@ -563,12 +568,21 @@ object Replicator {
*/
final case class Unsubscribe[A <: ReplicatedData](key: Key[A], subscriber: ActorRef) extends ReplicatorMessage
/**
* @see [[Replicator.Subscribe]]
*/
sealed trait SubscribeResponse[A <: ReplicatedData] extends NoSerializationVerificationNeeded {
def key: Key[A]
}
/**
* The data value is retrieved with [[#get]] using the typed key.
*
* @see [[Replicator.Subscribe]]
*/
final case class Changed[A <: ReplicatedData](key: Key[A])(data: A) extends ReplicatorMessage {
final case class Changed[A <: ReplicatedData](key: Key[A])(data: A)
extends SubscribeResponse[A]
with ReplicatorMessage {
/**
* The data value, with correct type.
@ -585,9 +599,10 @@ object Replicator {
def dataValue: A = data
}
final case class Deleted[A <: ReplicatedData](key: Key[A]) extends NoSerializationVerificationNeeded {
override def toString: String = s"Deleted [$key]"
}
/**
* @see [[Replicator.Subscribe]]
*/
final case class Deleted[A <: ReplicatedData](key: Key[A]) extends SubscribeResponse[A]
object Update {
@ -691,6 +706,11 @@ object Replicator {
*/
final case class UpdateTimeout[A <: ReplicatedData](key: Key[A], request: Option[Any]) extends UpdateFailure[A]
/**
* The [[Update]] couldn't be performed because the entry has been deleted.
*/
final case class UpdateDataDeleted[A <: ReplicatedData](key: Key[A], request: Option[Any]) extends UpdateResponse[A]
/**
* If the `modify` function of the [[Update]] throws an exception the reply message
* will be this `ModifyFailure` message. The original exception is included as `cause`.
@ -1172,7 +1192,8 @@ object Replicator {
*
* A deleted key cannot be reused again, but it is still recommended to delete unused
* data entries because that reduces the replication overhead when new nodes join the cluster.
* Subsequent `Delete`, `Update` and `Get` requests will be replied with [[Replicator.DataDeleted]].
* Subsequent `Delete`, `Update` and `Get` requests will be replied with [[Replicator.DataDeleted]],
* [[Replicator.UpdateDataDeleted]] and [[Replicator.GetDataDeleted]] respectively.
* Subscribers will receive [[Replicator.Deleted]].
*
* In the `Delete` message you can pass an optional request context in the same way as for the
@ -1515,10 +1536,10 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
def receiveGet(key: KeyR, consistency: ReadConsistency, req: Option[Any]): Unit = {
val localValue = getData(key.id)
log.debug("Received Get for key [{}]", key)
log.debug("Received Get for key [{}].", key)
if (isLocalGet(consistency)) {
val reply = localValue match {
case Some(DataEnvelope(DeletedData, _, _)) => DataDeleted(key, req)
case Some(DataEnvelope(DeletedData, _, _)) => GetDataDeleted(key, req)
case Some(DataEnvelope(data, _, _)) => GetSuccess(key, req)(data)
case None => NotFound(key, req)
}
@ -1559,7 +1580,8 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
Try {
localValue match {
case Some(DataEnvelope(DeletedData, _, _)) => throw new DataDeleted(key, req)
case Some(envelope @ DataEnvelope(DeletedData, _, _)) =>
(envelope, None)
case Some(envelope @ DataEnvelope(existing, _, _)) =>
modify(Some(existing)) match {
case d: DeltaReplicatedData if deltaCrdtEnabled =>
@ -1575,8 +1597,11 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
}
}
} match {
case Success((DataEnvelope(DeletedData, _, _), _)) =>
log.debug("Received Update for deleted key [{}].", key)
replyTo ! UpdateDataDeleted(key, req)
case Success((envelope, delta)) =>
log.debug("Received Update for key [{}]", key)
log.debug("Received Update for key [{}].", key)
// handle the delta
delta match {
@ -1628,9 +1653,6 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
Some(StoreReply(UpdateSuccess(key, req), StoreFailure(key, req), writeAggregator)))
}
}
case Failure(e: DataDeleted[_]) =>
log.debug("Received Update for deleted key [{}]", key)
replyTo ! e
case Failure(e) =>
log.debug("Received Update for key [{}], failed: {}", key, e.getMessage)
replyTo ! ModifyFailure(key, "Update failed: " + e.getMessage, e, req)
@ -1859,7 +1881,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
val isDebugEnabled = log.isDebugEnabled
if (isDebugEnabled)
log.debug(
"Received DeltaPropagation from [{}], containing [{}]",
"Received DeltaPropagation from [{}], containing [{}].",
fromNode.address,
deltas.collect { case (key, Delta(_, fromSeqNr, toSeqNr)) => s"$key $fromSeqNr-$toSeqNr" }.mkString(", "))
@ -1960,7 +1982,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
def receiveStatus(otherDigests: Map[KeyId, Digest], chunk: Int, totChunks: Int, fromSystemUid: Option[Long]): Unit = {
if (log.isDebugEnabled)
log.debug(
"Received gossip status from [{}], chunk [{}] of [{}] containing [{}]",
"Received gossip status from [{}], chunk [{}] of [{}] containing [{}].",
replyTo.path.address,
(chunk + 1),
totChunks,
@ -2008,7 +2030,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
def receiveGossip(updatedData: Map[KeyId, DataEnvelope], sendBack: Boolean, fromSystemUid: Option[Long]): Unit = {
if (log.isDebugEnabled)
log.debug("Received gossip from [{}], containing [{}]", replyTo.path.address, updatedData.keys.mkString(", "))
log.debug("Received gossip from [{}], containing [{}].", replyTo.path.address, updatedData.keys.mkString(", "))
var replyData = Map.empty[KeyId, DataEnvelope]
updatedData.foreach {
case (key, envelope) =>
@ -2545,7 +2567,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
def waitReadRepairAck(envelope: Replicator.Internal.DataEnvelope): Receive = {
case ReadRepairAck =>
val replyMsg =
if (envelope.data == DeletedData) DataDeleted(key, req)
if (envelope.data == DeletedData) GetDataDeleted(key, req)
else GetSuccess(key, req)(envelope.data)
replyTo.tell(replyMsg, context.parent)
context.stop(self)