diff --git a/akka-distributed-data/src/main/java/akka/cluster/ddata/protobuf/msg/ReplicatorMessages.java b/akka-distributed-data/src/main/java/akka/cluster/ddata/protobuf/msg/ReplicatorMessages.java
index 63c51ac058..d4e0dcde92 100644
--- a/akka-distributed-data/src/main/java/akka/cluster/ddata/protobuf/msg/ReplicatorMessages.java
+++ b/akka-distributed-data/src/main/java/akka/cluster/ddata/protobuf/msg/ReplicatorMessages.java
@@ -12721,6 +12721,24 @@ public final class ReplicatorMessages {
      */
     akka.cluster.ddata.protobuf.msg.ReplicatorMessages.DeltaPropagation.EntryOrBuilder getEntriesOrBuilder(
         int index);
+
+    // optional bool reply = 3;
+    /**
+     * <code>optional bool reply = 3;</code>
+     *
+     * <pre>
+     * no reply if not set 
+     * </pre>
+     */
+    boolean hasReply();
+    /**
+     * <code>optional bool reply = 3;</code>
+     *
+     * <pre>
+     * no reply if not set 
+     * </pre>
+     */
+    boolean getReply();
   }
   /**
    * Protobuf type {@code akka.cluster.ddata.DeltaPropagation}
@@ -12794,6 +12812,11 @@ public final class ReplicatorMessages {
               entries_.add(input.readMessage(akka.cluster.ddata.protobuf.msg.ReplicatorMessages.DeltaPropagation.Entry.PARSER, extensionRegistry));
               break;
             }
+            case 24: {
+              bitField0_ |= 0x00000002;
+              reply_ = input.readBool();
+              break;
+            }
           }
         }
       } catch (akka.protobuf.InvalidProtocolBufferException e) {
@@ -13784,9 +13807,34 @@ public final class ReplicatorMessages {
       return entries_.get(index);
     }
 
+    // optional bool reply = 3;
+    public static final int REPLY_FIELD_NUMBER = 3;
+    private boolean reply_;
+    /**
+     * <code>optional bool reply = 3;</code>
+     *
+     * <pre>
+     * no reply if not set 
+     * </pre>
+     */
+    public boolean hasReply() {
+      return ((bitField0_ & 0x00000002) == 0x00000002);
+    }
+    /**
+     * <code>optional bool reply = 3;</code>
+     *
+     * <pre>
+     * no reply if not set 
+     * </pre>
+     */
+    public boolean getReply() {
+      return reply_;
+    }
+
     private void initFields() {
       fromNode_ = akka.cluster.ddata.protobuf.msg.ReplicatorMessages.UniqueAddress.getDefaultInstance();
       entries_ = java.util.Collections.emptyList();
+      reply_ = false;
     }
     private byte memoizedIsInitialized = -1;
     public final boolean isInitialized() {
@@ -13820,6 +13868,9 @@ public final class ReplicatorMessages {
       for (int i = 0; i < entries_.size(); i++) {
         output.writeMessage(2, entries_.get(i));
       }
+      if (((bitField0_ & 0x00000002) == 0x00000002)) {
+        output.writeBool(3, reply_);
+      }
       getUnknownFields().writeTo(output);
     }
 
@@ -13837,6 +13888,10 @@ public final class ReplicatorMessages {
         size += akka.protobuf.CodedOutputStream
           .computeMessageSize(2, entries_.get(i));
       }
+      if (((bitField0_ & 0x00000002) == 0x00000002)) {
+        size += akka.protobuf.CodedOutputStream
+          .computeBoolSize(3, reply_);
+      }
       size += getUnknownFields().getSerializedSize();
       memoizedSerializedSize = size;
       return size;
@@ -13967,6 +14022,8 @@ public final class ReplicatorMessages {
         } else {
           entriesBuilder_.clear();
         }
+        reply_ = false;
+        bitField0_ = (bitField0_ & ~0x00000004);
         return this;
       }
 
@@ -14012,6 +14069,10 @@ public final class ReplicatorMessages {
         } else {
           result.entries_ = entriesBuilder_.build();
         }
+        if (((from_bitField0_ & 0x00000004) == 0x00000004)) {
+          to_bitField0_ |= 0x00000002;
+        }
+        result.reply_ = reply_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -14057,6 +14118,9 @@ public final class ReplicatorMessages {
             }
           }
         }
+        if (other.hasReply()) {
+          setReply(other.getReply());
+        }
         this.mergeUnknownFields(other.getUnknownFields());
         return this;
       }
@@ -14455,6 +14519,55 @@ public final class ReplicatorMessages {
         return entriesBuilder_;
       }
 
+      // optional bool reply = 3;
+      private boolean reply_ ;
+      /**
+       * <code>optional bool reply = 3;</code>
+       *
+       * <pre>
+       * no reply if not set 
+       * </pre>
+       */
+      public boolean hasReply() {
+        return ((bitField0_ & 0x00000004) == 0x00000004);
+      }
+      /**
+       * <code>optional bool reply = 3;</code>
+       *
+       * <pre>
+       * no reply if not set 
+       * </pre>
+       */
+      public boolean getReply() {
+        return reply_;
+      }
+      /**
+       * <code>optional bool reply = 3;</code>
+       *
+       * <pre>
+       * no reply if not set 
+       * </pre>
+       */
+      public Builder setReply(boolean value) {
+        bitField0_ |= 0x00000004;
+        reply_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional bool reply = 3;</code>
+       *
+       * <pre>
+       * no reply if not set 
+       * </pre>
+       */
+      public Builder clearReply() {
+        bitField0_ = (bitField0_ & ~0x00000004);
+        reply_ = false;
+        onChanged();
+        return this;
+      }
+
       // @@protoc_insertion_point(builder_scope:akka.cluster.ddata.DeltaPropagation)
     }
 
@@ -19212,27 +19325,28 @@ public final class ReplicatorMessages {
       " \002(\010\0221\n\007entries\030\002 \003(\0132 .akka.cluster.dda",
       "ta.Gossip.Entry\032H\n\005Entry\022\013\n\003key\030\001 \002(\t\0222\n" +
       "\010envelope\030\002 \002(\0132 .akka.cluster.ddata.Dat" +
-      "aEnvelope\"\362\001\n\020DeltaPropagation\0223\n\010fromNo" +
+      "aEnvelope\"\201\002\n\020DeltaPropagation\0223\n\010fromNo" +
       "de\030\001 \002(\0132!.akka.cluster.ddata.UniqueAddr" +
       "ess\022;\n\007entries\030\002 \003(\0132*.akka.cluster.ddat" +
-      "a.DeltaPropagation.Entry\032l\n\005Entry\022\013\n\003key" +
-      "\030\001 \002(\t\0222\n\010envelope\030\002 \002(\0132 .akka.cluster." +
-      "ddata.DataEnvelope\022\021\n\tfromSeqNr\030\003 \002(\003\022\017\n" +
-      "\007toSeqNr\030\004 \001(\003\"X\n\rUniqueAddress\022,\n\007addre" +
-      "ss\030\001 \002(\0132\033.akka.cluster.ddata.Address\022\013\n",
-      "\003uid\030\002 \002(\017\022\014\n\004uid2\030\003 \001(\017\")\n\007Address\022\020\n\010h" +
-      "ostname\030\001 \002(\t\022\014\n\004port\030\002 \002(\r\"\224\001\n\rVersionV" +
-      "ector\0228\n\007entries\030\001 \003(\0132\'.akka.cluster.dd" +
-      "ata.VersionVector.Entry\032I\n\005Entry\022/\n\004node" +
-      "\030\001 \002(\0132!.akka.cluster.ddata.UniqueAddres" +
-      "s\022\017\n\007version\030\002 \002(\003\"V\n\014OtherMessage\022\027\n\017en" +
-      "closedMessage\030\001 \002(\014\022\024\n\014serializerId\030\002 \002(" +
-      "\005\022\027\n\017messageManifest\030\004 \001(\014\"\036\n\nStringGSet" +
-      "\022\020\n\010elements\030\001 \003(\t\"\205\001\n\023DurableDataEnvelo" +
-      "pe\022.\n\004data\030\001 \002(\0132 .akka.cluster.ddata.Ot",
-      "herMessage\022>\n\007pruning\030\002 \003(\0132-.akka.clust" +
-      "er.ddata.DataEnvelope.PruningEntryB#\n\037ak" +
-      "ka.cluster.ddata.protobuf.msgH\001"
+      "a.DeltaPropagation.Entry\022\r\n\005reply\030\003 \001(\010\032" +
+      "l\n\005Entry\022\013\n\003key\030\001 \002(\t\0222\n\010envelope\030\002 \002(\0132" +
+      " .akka.cluster.ddata.DataEnvelope\022\021\n\tfro" +
+      "mSeqNr\030\003 \002(\003\022\017\n\007toSeqNr\030\004 \001(\003\"X\n\rUniqueA" +
+      "ddress\022,\n\007address\030\001 \002(\0132\033.akka.cluster.d",
+      "data.Address\022\013\n\003uid\030\002 \002(\017\022\014\n\004uid2\030\003 \001(\017\"" +
+      ")\n\007Address\022\020\n\010hostname\030\001 \002(\t\022\014\n\004port\030\002 \002" +
+      "(\r\"\224\001\n\rVersionVector\0228\n\007entries\030\001 \003(\0132\'." +
+      "akka.cluster.ddata.VersionVector.Entry\032I" +
+      "\n\005Entry\022/\n\004node\030\001 \002(\0132!.akka.cluster.dda" +
+      "ta.UniqueAddress\022\017\n\007version\030\002 \002(\003\"V\n\014Oth" +
+      "erMessage\022\027\n\017enclosedMessage\030\001 \002(\014\022\024\n\014se" +
+      "rializerId\030\002 \002(\005\022\027\n\017messageManifest\030\004 \001(" +
+      "\014\"\036\n\nStringGSet\022\020\n\010elements\030\001 \003(\t\"\205\001\n\023Du" +
+      "rableDataEnvelope\022.\n\004data\030\001 \002(\0132 .akka.c",
+      "luster.ddata.OtherMessage\022>\n\007pruning\030\002 \003" +
+      "(\0132-.akka.cluster.ddata.DataEnvelope.Pru" +
+      "ningEntryB#\n\037akka.cluster.ddata.protobuf" +
+      ".msgH\001"
     };
     akka.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
       new akka.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -19346,7 +19460,7 @@ public final class ReplicatorMessages {
           internal_static_akka_cluster_ddata_DeltaPropagation_fieldAccessorTable = new
             akka.protobuf.GeneratedMessage.FieldAccessorTable(
               internal_static_akka_cluster_ddata_DeltaPropagation_descriptor,
-              new java.lang.String[] { "FromNode", "Entries", });
+              new java.lang.String[] { "FromNode", "Entries", "Reply", });
           internal_static_akka_cluster_ddata_DeltaPropagation_Entry_descriptor =
             internal_static_akka_cluster_ddata_DeltaPropagation_descriptor.getNestedTypes().get(0);
           internal_static_akka_cluster_ddata_DeltaPropagation_Entry_fieldAccessorTable = new
diff --git a/akka-distributed-data/src/main/protobuf/ReplicatorMessages.proto b/akka-distributed-data/src/main/protobuf/ReplicatorMessages.proto
index 0582fac91b..0c83967b07 100644
--- a/akka-distributed-data/src/main/protobuf/ReplicatorMessages.proto
+++ b/akka-distributed-data/src/main/protobuf/ReplicatorMessages.proto
@@ -107,6 +107,7 @@ message DeltaPropagation {
   
   required UniqueAddress fromNode = 1;
   repeated Entry entries = 2;
+  optional bool reply = 3; // no reply if not set 
 }
 
 message UniqueAddress {
diff --git a/akka-distributed-data/src/main/scala/akka/cluster/ddata/Replicator.scala b/akka-distributed-data/src/main/scala/akka/cluster/ddata/Replicator.scala
index 0a96101ad0..5e85fa294e 100644
--- a/akka-distributed-data/src/main/scala/akka/cluster/ddata/Replicator.scala
+++ b/akka-distributed-data/src/main/scala/akka/cluster/ddata/Replicator.scala
@@ -743,7 +743,8 @@ object Replicator {
     final case class Gossip(updatedData: Map[KeyId, DataEnvelope], sendBack: Boolean) extends ReplicatorMessage
 
     final case class Delta(dataEnvelope: DataEnvelope, fromSeqNr: Long, toSeqNr: Long)
-    final case class DeltaPropagation(fromNode: UniqueAddress, deltas: Map[KeyId, Delta]) extends ReplicatorMessage
+    final case class DeltaPropagation(fromNode: UniqueAddress, reply: Boolean, deltas: Map[KeyId, Delta]) extends ReplicatorMessage
+    case object DeltaNack extends ReplicatorMessage with DeadLetterSuppression
 
   }
 }
@@ -990,7 +991,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
     override def createDeltaPropagation(deltas: Map[KeyId, (ReplicatedData, Long, Long)]): DeltaPropagation = {
       // Important to include the pruning state in the deltas. For example if the delta is based
       // on an entry that has been pruned but that has not yet been performed on the target node.
-      DeltaPropagation(selfUniqueAddress, deltas.map {
+      DeltaPropagation(selfUniqueAddress, reply = false, deltas.map {
         case (key, (d, fromSeqNr, toSeqNr)) ⇒ getData(key) match {
           case Some(envelope) ⇒ key → Delta(envelope.copy(data = d), fromSeqNr, toSeqNr)
           case None           ⇒ key → Delta(DataEnvelope(d), fromSeqNr, toSeqNr)
@@ -1149,7 +1150,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
     case Read(key)                              ⇒ receiveRead(key)
     case Write(key, envelope)                   ⇒ receiveWrite(key, envelope)
     case ReadRepair(key, envelope)              ⇒ receiveReadRepair(key, envelope)
-    case DeltaPropagation(from, deltas)         ⇒ receiveDeltaPropagation(from, deltas)
+    case DeltaPropagation(from, reply, deltas)  ⇒ receiveDeltaPropagation(from, reply, deltas)
     case FlushChanges                           ⇒ receiveFlushChanges()
     case DeltaPropagationTick                   ⇒ receiveDeltaPropagationTick()
     case GossipTick                             ⇒ receiveGossipTick()
@@ -1242,13 +1243,16 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
           else
             replyTo ! UpdateSuccess(key, req)
         } else {
-          val writeEnvelope = delta match {
-            case Some(d: RequiresCausalDeliveryOfDeltas) ⇒ newEnvelope
-            case Some(d)                                 ⇒ DataEnvelope(d)
-            case None                                    ⇒ newEnvelope
+          val (writeEnvelope, writeDelta) = delta match {
+            case Some(d: RequiresCausalDeliveryOfDeltas) ⇒
+              val v = deltaPropagationSelector.currentVersion(key.id)
+              (newEnvelope, Some(Delta(newEnvelope.copy(data = d), v, v)))
+            case Some(d) ⇒ (newEnvelope.copy(data = d), None)
+            case None    ⇒ (newEnvelope, None)
           }
           val writeAggregator =
-            context.actorOf(WriteAggregator.props(key, writeEnvelope, writeConsistency, req, nodes, unreachable, replyTo, durable)
+            context.actorOf(WriteAggregator.props(key, writeEnvelope, writeDelta, writeConsistency,
+              req, nodes, unreachable, replyTo, durable)
               .withDispatcher(context.props.dispatcher))
           if (durable) {
             durableStore ! Store(key.id, new DurableDataEnvelope(newEnvelope),
@@ -1274,14 +1278,20 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
       case _                              ⇒ false
     }
 
-  def receiveWrite(key: KeyId, envelope: DataEnvelope): Unit = {
-    write(key, envelope) match {
+  def receiveWrite(key: KeyId, envelope: DataEnvelope): Unit =
+    writeAndStore(key, envelope, reply = true)
+
+  def writeAndStore(key: KeyId, writeEnvelope: DataEnvelope, reply: Boolean): Unit = {
+    write(key, writeEnvelope) match {
       case Some(newEnvelope) ⇒
-        if (isDurable(key))
-          durableStore ! Store(key, new DurableDataEnvelope(newEnvelope), Some(StoreReply(WriteAck, WriteNack, replyTo)))
-        else
+        if (isDurable(key)) {
+          val storeReply = if (reply) Some(StoreReply(WriteAck, WriteNack, replyTo)) else None
+          durableStore ! Store(key, new DurableDataEnvelope(newEnvelope), storeReply)
+        } else if (reply)
           replyTo ! WriteAck
       case None ⇒
+        if (reply)
+          replyTo ! WriteNack
     }
   }
 
@@ -1316,17 +1326,8 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
     }
   }
 
-  def writeAndStore(key: KeyId, writeEnvelope: DataEnvelope): Unit = {
-    write(key, writeEnvelope) match {
-      case Some(newEnvelope) ⇒
-        if (isDurable(key))
-          durableStore ! Store(key, new DurableDataEnvelope(newEnvelope), None)
-      case None ⇒
-    }
-  }
-
   def receiveReadRepair(key: KeyId, writeEnvelope: DataEnvelope): Unit = {
-    writeAndStore(key, writeEnvelope)
+    writeAndStore(key, writeEnvelope, reply = false)
     replyTo ! ReadRepairAck
   }
 
@@ -1353,7 +1354,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
             replyTo ! DeleteSuccess(key, req)
         } else {
           val writeAggregator =
-            context.actorOf(WriteAggregator.props(key, DeletedEnvelope, consistency, req, nodes, unreachable, replyTo, durable)
+            context.actorOf(WriteAggregator.props(key, DeletedEnvelope, None, consistency, req, nodes, unreachable, replyTo, durable)
               .withDispatcher(context.props.dispatcher))
           if (durable) {
             durableStore ! Store(key.id, new DurableDataEnvelope(DeletedEnvelope),
@@ -1464,7 +1465,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
       deltaPropagationSelector.cleanupDeltaEntries()
   }
 
-  def receiveDeltaPropagation(fromNode: UniqueAddress, deltas: Map[KeyId, Delta]): Unit =
+  def receiveDeltaPropagation(fromNode: UniqueAddress, reply: Boolean, deltas: Map[KeyId, Delta]): Unit =
     if (deltaCrdtEnabled) {
       try {
         val isDebugEnabled = log.isDebugEnabled
@@ -1485,20 +1486,22 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
                 if (isDebugEnabled) log.debug(
                   "Skipping DeltaPropagation from [{}] for [{}] because toSeqNr [{}] already handled [{}]",
                   fromNode.address, key, toSeqNr, currentSeqNr)
+                if (reply) replyTo ! WriteAck
               } else if (fromSeqNr > (currentSeqNr + 1)) {
                 if (isDebugEnabled) log.debug(
                   "Skipping DeltaPropagation from [{}] for [{}] because missing deltas between [{}-{}]",
                   fromNode.address, key, currentSeqNr + 1, fromSeqNr - 1)
+                if (reply) replyTo ! DeltaNack
               } else {
                 if (isDebugEnabled) log.debug(
                   "Applying DeltaPropagation from [{}] for [{}] with sequence numbers [{}], current was [{}]",
                   fromNode.address, key, s"$fromSeqNr-$toSeqNr", currentSeqNr)
                 val newEnvelope = envelope.copy(deltaVersions = VersionVector(fromNode, toSeqNr))
-                writeAndStore(key, newEnvelope)
+                writeAndStore(key, newEnvelope, reply)
               }
             case (key, Delta(envelope, _, _)) ⇒
               // causal delivery of deltas not needed, just apply it
-              writeAndStore(key, envelope)
+              writeAndStore(key, envelope, reply)
           }
         }
       } catch {
@@ -1507,6 +1510,9 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
           // mixing nodes with incompatible delta-CRDT types
           log.warning("Couldn't process DeltaPropagation from [] due to {}", fromNode, e)
       }
+    } else {
+      // !deltaCrdtEnabled
+      if (reply) replyTo ! DeltaNack
     }
 
   def receiveGossipTick(): Unit = {
@@ -1583,7 +1589,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
     updatedData.foreach {
       case (key, envelope) ⇒
         val hadData = dataEntries.contains(key)
-        writeAndStore(key, envelope)
+        writeAndStore(key, envelope, reply = false)
         if (sendBack) getData(key) match {
           case Some(d) ⇒
             if (hadData || d.pruning.nonEmpty)
@@ -1842,13 +1848,14 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
   def props(
     key:         KeyR,
     envelope:    Replicator.Internal.DataEnvelope,
+    delta:       Option[Replicator.Internal.Delta],
     consistency: Replicator.WriteConsistency,
     req:         Option[Any],
     nodes:       Set[Address],
     unreachable: Set[Address],
     replyTo:     ActorRef,
     durable:     Boolean): Props =
-    Props(new WriteAggregator(key, envelope, consistency, req, nodes, unreachable, replyTo, durable))
+    Props(new WriteAggregator(key, envelope, delta, consistency, req, nodes, unreachable, replyTo, durable))
       .withDeploy(Deploy.local)
 }
 
@@ -1858,6 +1865,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
 @InternalApi private[akka] class WriteAggregator(
   key:                      KeyR,
   envelope:                 Replicator.Internal.DataEnvelope,
+  delta:                    Option[Replicator.Internal.Delta],
   consistency:              Replicator.WriteConsistency,
   req:                      Option[Any],
   override val nodes:       Set[Address],
@@ -1869,6 +1877,8 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
   import Replicator.Internal._
   import ReadWriteAggregator._
 
+  val selfUniqueAddress = Cluster(context.system).selfUniqueAddress
+
   override def timeout: FiniteDuration = consistency.timeout
 
   override val doneWhenRemainingSize = consistency match {
@@ -1883,12 +1893,20 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
   }
 
   val writeMsg = Write(key.id, envelope)
+  val deltaMsg = delta match {
+    case None    ⇒ None
+    case Some(d) ⇒ Some(DeltaPropagation(selfUniqueAddress, reply = true, Map(key.id → d)))
+  }
 
   var gotLocalStoreReply = !durable
   var gotWriteNackFrom = Set.empty[Address]
 
   override def preStart(): Unit = {
-    primaryNodes.foreach { replica(_) ! writeMsg }
+    val msg = deltaMsg match {
+      case Some(d) ⇒ d
+      case None    ⇒ writeMsg
+    }
+    primaryNodes.foreach { replica(_) ! msg }
 
     if (isDone) reply(isTimeout = false)
   }
@@ -1900,16 +1918,26 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
     case WriteNack ⇒
       gotWriteNackFrom += senderAddress()
       if (isDone) reply(isTimeout = false)
+    case DeltaNack ⇒
+    // ok, will be retried with full state
 
     case _: Replicator.UpdateSuccess[_] ⇒
       gotLocalStoreReply = true
       if (isDone) reply(isTimeout = false)
     case f: Replicator.StoreFailure[_] ⇒
       gotLocalStoreReply = true
-      gotWriteNackFrom += Cluster(context.system).selfAddress
+      gotWriteNackFrom += selfUniqueAddress.address
       if (isDone) reply(isTimeout = false)
 
     case SendToSecondary ⇒
+      deltaMsg match {
+        case None ⇒
+        case Some(d) ⇒
+          // Deltas must be applied in order and we can't keep track of ordering of
+          // simultaneous updates so there is a chance that the delta could not be applied.
+          // Try again with the full state to the primary nodes that have not acked.
+          primaryNodes.toSet.intersect(remaining).foreach { replica(_) ! writeMsg }
+      }
       secondaryNodes.foreach { replica(_) ! writeMsg }
     case ReceiveTimeout ⇒
       reply(isTimeout = true)
@@ -1934,7 +1962,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
       if (isSuccess && isDelete) DeleteSuccess(key, req)
       else if (isSuccess) UpdateSuccess(key, req)
       else if (isTimeoutOrNotEnoughNodes && isDelete) ReplicationDeleteFailure(key, req)
-      else if (isTimeoutOrNotEnoughNodes) UpdateTimeout(key, req)
+      else if (isTimeoutOrNotEnoughNodes || !durable) UpdateTimeout(key, req)
       else StoreFailure(key, req)
 
     replyTo.tell(replyMsg, context.parent)
diff --git a/akka-distributed-data/src/main/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializer.scala b/akka-distributed-data/src/main/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializer.scala
index c48b8f4750..d24682884f 100644
--- a/akka-distributed-data/src/main/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializer.scala
+++ b/akka-distributed-data/src/main/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializer.scala
@@ -176,6 +176,7 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
   val WriteNackManifest = "O"
   val DurableDataEnvelopeManifest = "P"
   val DeltaPropagationManifest = "Q"
+  val DeltaNackManifest = "R"
 
   private val fromBinaryMap = collection.immutable.HashMap[String, Array[Byte] ⇒ AnyRef](
     GetManifest → getFromBinary,
@@ -194,6 +195,7 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
     GossipManifest → gossipFromBinary,
     DeltaPropagationManifest → deltaPropagationFromBinary,
     WriteNackManifest → (_ ⇒ WriteNack),
+    DeltaNackManifest → (_ ⇒ DeltaNack),
     DurableDataEnvelopeManifest → durableDataEnvelopeFromBinary)
 
   override def manifest(obj: AnyRef): String = obj match {
@@ -214,6 +216,7 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
     case _: Unsubscribe[_]      ⇒ UnsubscribeManifest
     case _: Gossip              ⇒ GossipManifest
     case WriteNack              ⇒ WriteNackManifest
+    case DeltaNack              ⇒ DeltaNackManifest
     case _ ⇒
       throw new IllegalArgumentException(s"Can't serialize object of type ${obj.getClass} in [${getClass.getName}]")
   }
@@ -236,6 +239,7 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
     case m: Unsubscribe[_]      ⇒ unsubscribeToProto(m).toByteArray
     case m: Gossip              ⇒ compress(gossipToProto(m))
     case WriteNack              ⇒ dm.Empty.getDefaultInstance.toByteArray
+    case DeltaNack              ⇒ dm.Empty.getDefaultInstance.toByteArray
     case _ ⇒
       throw new IllegalArgumentException(s"Can't serialize object of type ${obj.getClass} in [${getClass.getName}]")
   }
@@ -289,6 +293,8 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
   private def deltaPropagationToProto(deltaPropagation: DeltaPropagation): dm.DeltaPropagation = {
     val b = dm.DeltaPropagation.newBuilder()
       .setFromNode(uniqueAddressToProto(deltaPropagation.fromNode))
+    if (deltaPropagation.reply)
+      b.setReply(deltaPropagation.reply)
     val entries = deltaPropagation.deltas.foreach {
       case (key, Delta(data, fromSeqNr, toSeqNr)) ⇒
         val b2 = dm.DeltaPropagation.Entry.newBuilder()
@@ -304,8 +310,10 @@ class ReplicatorMessageSerializer(val system: ExtendedActorSystem)
 
   private def deltaPropagationFromBinary(bytes: Array[Byte]): DeltaPropagation = {
     val deltaPropagation = dm.DeltaPropagation.parseFrom(bytes)
+    val reply = deltaPropagation.hasReply && deltaPropagation.getReply
     DeltaPropagation(
       uniqueAddressFromProto(deltaPropagation.getFromNode),
+      reply,
       deltaPropagation.getEntriesList.asScala.map { e ⇒
         val fromSeqNr = e.getFromSeqNr
         val toSeqNr = if (e.hasToSeqNr) e.getToSeqNr else fromSeqNr
diff --git a/akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorDeltaSpec.scala b/akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorDeltaSpec.scala
index 44910f3afd..c1df7bffdb 100644
--- a/akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorDeltaSpec.scala
+++ b/akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorDeltaSpec.scala
@@ -22,7 +22,7 @@ object ReplicatorDeltaSpec extends MultiNodeConfig {
   val fourth = role("fourth")
 
   commonConfig(ConfigFactory.parseString("""
-    akka.loglevel = DEBUG
+    akka.loglevel = INFO
     akka.actor.provider = "cluster"
     akka.log-dead-letters-during-shutdown = off
     """))
@@ -129,6 +129,8 @@ class ReplicatorDeltaSpec extends MultiNodeSpec(ReplicatorDeltaSpec) with STMult
     r
   }
 
+  val writeAll = WriteAll(5.seconds)
+
   var afterCounter = 0
   def enterBarrierAfterTestStep(): Unit = {
     afterCounter += 1
@@ -197,7 +199,7 @@ class ReplicatorDeltaSpec extends MultiNodeSpec(ReplicatorDeltaSpec) with STMult
         awaitAssert {
           val p = TestProbe()
           List(KeyD, KeyE, KeyF).foreach { key ⇒
-            fullStateReplicator.tell(Get(key, ReadLocal), p.ref)
+            deltaReplicator.tell(Get(key, ReadLocal), p.ref)
             p.expectMsgType[GetSuccess[ORSet[String]]].dataValue.elements should ===(Set("a"))
           }
         }
@@ -206,6 +208,48 @@ class ReplicatorDeltaSpec extends MultiNodeSpec(ReplicatorDeltaSpec) with STMult
       enterBarrierAfterTestStep()
     }
 
+    "work with write consistency" in {
+      runOn(first) {
+        val p1 = TestProbe()
+        fullStateReplicator.tell(Update(KeyD, ORSet.empty[String], writeAll)(_ + "A"), p1.ref)
+        deltaReplicator.tell(Update(KeyD, ORSet.empty[String], writeAll)(_ + "A"), p1.ref)
+        p1.expectMsgType[UpdateSuccess[_]]
+        p1.expectMsgType[UpdateSuccess[_]]
+      }
+      enterBarrier("write-1")
+
+      val p = TestProbe()
+      deltaReplicator.tell(Get(KeyD, ReadLocal), p.ref)
+      p.expectMsgType[GetSuccess[ORSet[String]]].dataValue.elements should ===(Set("a", "A"))
+
+      // and also when doing several at the same time (deltas may be reordered) and then we
+      // retry with full state to sort it out
+      runOn(first) {
+        val p1 = TestProbe()
+        deltaReplicator.tell(Update(KeyD, ORSet.empty[String], writeAll)(_ + "B"), p1.ref)
+        deltaReplicator.tell(Update(KeyD, ORSet.empty[String], writeAll)(_ + "C"), p1.ref)
+        deltaReplicator.tell(Update(KeyD, ORSet.empty[String], writeAll)(_ + "D"), p1.ref)
+        p1.expectMsgType[UpdateSuccess[_]]
+        p1.expectMsgType[UpdateSuccess[_]]
+        p1.expectMsgType[UpdateSuccess[_]]
+      }
+      enterBarrier("write-2")
+      deltaReplicator.tell(Get(KeyD, ReadLocal), p.ref)
+      p.expectMsgType[GetSuccess[ORSet[String]]].dataValue.elements should ===(Set("a", "A", "B", "C", "D"))
+
+      // add same to the fullStateReplicator so they are in sync
+      runOn(first) {
+        val p1 = TestProbe()
+        fullStateReplicator.tell(Update(KeyD, ORSet.empty[String], writeAll)(_ + "A" + "B" + "C" + "D"), p1.ref)
+        p1.expectMsgType[UpdateSuccess[_]]
+      }
+      enterBarrier("write-3")
+      fullStateReplicator.tell(Get(KeyD, ReadLocal), p.ref)
+      p.expectMsgType[GetSuccess[ORSet[String]]].dataValue.elements should ===(Set("a", "A", "B", "C", "D"))
+
+      enterBarrierAfterTestStep()
+    }
+
     "be eventually consistent" in {
       val operations = generateOperations(onNode = myself)
       log.debug(s"random operations on [${myself.name}]: ${operations.mkString(", ")}")
diff --git a/akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorORSetDeltaSpec.scala b/akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorORSetDeltaSpec.scala
index 04232f3f82..3d1a9581a8 100644
--- a/akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorORSetDeltaSpec.scala
+++ b/akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorORSetDeltaSpec.scala
@@ -19,7 +19,7 @@ object ReplicatorORSetDeltaSpec extends MultiNodeConfig {
   val third = role("third")
 
   commonConfig(ConfigFactory.parseString("""
-    akka.loglevel = DEBUG
+    akka.loglevel = INFO
     akka.actor.provider = "cluster"
     akka.log-dead-letters-during-shutdown = off
     """))
diff --git a/akka-distributed-data/src/test/scala/akka/cluster/ddata/DeltaPropagationSelectorSpec.scala b/akka-distributed-data/src/test/scala/akka/cluster/ddata/DeltaPropagationSelectorSpec.scala
index e511e024fa..7dd1fea309 100644
--- a/akka-distributed-data/src/test/scala/akka/cluster/ddata/DeltaPropagationSelectorSpec.scala
+++ b/akka-distributed-data/src/test/scala/akka/cluster/ddata/DeltaPropagationSelectorSpec.scala
@@ -19,7 +19,7 @@ object DeltaPropagationSelectorSpec {
     override val allNodes: Vector[Address]) extends DeltaPropagationSelector {
     override val gossipIntervalDivisor = 5
     override def createDeltaPropagation(deltas: Map[KeyId, (ReplicatedData, Long, Long)]): DeltaPropagation =
-      DeltaPropagation(selfUniqueAddress, deltas.mapValues {
+      DeltaPropagation(selfUniqueAddress, false, deltas.mapValues {
         case (d, fromSeqNr, toSeqNr) ⇒ Delta(DataEnvelope(d), fromSeqNr, toSeqNr)
       })
   }
@@ -50,7 +50,7 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
       selector.cleanupDeltaEntries()
       selector.hasDeltaEntries("A") should ===(true)
       selector.hasDeltaEntries("B") should ===(true)
-      val expected = DeltaPropagation(selfUniqueAddress, Map(
+      val expected = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(deltaA), 1L, 1L),
         "B" → Delta(DataEnvelope(deltaB), 1L, 1L)))
       selector.collectPropagations() should ===(Map(nodes(0) → expected))
@@ -64,7 +64,7 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
       val selector = new TestSelector(selfUniqueAddress, nodes.take(3))
       selector.update("A", deltaA)
       selector.update("B", deltaB)
-      val expected = DeltaPropagation(selfUniqueAddress, Map(
+      val expected = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(deltaA), 1L, 1L),
         "B" → Delta(DataEnvelope(deltaB), 1L, 1L)))
       selector.collectPropagations() should ===(Map(nodes(0) → expected, nodes(1) → expected))
@@ -82,17 +82,17 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
       val selector = new TestSelector(selfUniqueAddress, nodes.take(3))
       selector.update("A", deltaA)
       selector.update("B", deltaB)
-      val expected1 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected1 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(deltaA), 1L, 1L),
         "B" → Delta(DataEnvelope(deltaB), 1L, 1L)))
       selector.collectPropagations() should ===(Map(nodes(0) → expected1, nodes(1) → expected1))
       // new update before previous was propagated to all nodes
       selector.update("C", deltaC)
-      val expected2 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected2 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(deltaA), 1L, 1L),
         "B" → Delta(DataEnvelope(deltaB), 1L, 1L),
         "C" → Delta(DataEnvelope(deltaC), 1L, 1L)))
-      val expected3 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected3 = DeltaPropagation(selfUniqueAddress, false, Map(
         "C" → Delta(DataEnvelope(deltaC), 1L, 1L)))
       selector.collectPropagations() should ===(Map(nodes(2) → expected2, nodes(0) → expected3))
       selector.cleanupDeltaEntries()
@@ -114,12 +114,12 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
       selector.currentVersion("A") should ===(1L)
       selector.update("A", delta2)
       selector.currentVersion("A") should ===(2L)
-      val expected1 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected1 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(delta1.merge(delta2)), 1L, 2L)))
       selector.collectPropagations() should ===(Map(nodes(0) → expected1))
       selector.update("A", delta3)
       selector.currentVersion("A") should ===(3L)
-      val expected2 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected2 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(delta3), 3L, 3L)))
       selector.collectPropagations() should ===(Map(nodes(0) → expected2))
       selector.collectPropagations() should ===(Map.empty[Address, DeltaPropagation])
@@ -133,25 +133,25 @@ class DeltaPropagationSelectorSpec extends WordSpec with Matchers with TypeCheck
         override def nodesSliceSize(allNodesSize: Int): Int = 1
       }
       selector.update("A", delta1)
-      val expected1 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected1 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(delta1), 1L, 1L)))
       selector.collectPropagations() should ===(Map(nodes(0) → expected1))
 
       selector.update("A", delta2)
-      val expected2 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected2 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(delta1.merge(delta2)), 1L, 2L)))
       selector.collectPropagations() should ===(Map(nodes(1) → expected2))
 
       selector.update("A", delta3)
-      val expected3 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected3 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(delta1.merge(delta2).merge(delta3)), 1L, 3L)))
       selector.collectPropagations() should ===(Map(nodes(2) → expected3))
 
-      val expected4 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected4 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(delta2.merge(delta3)), 2L, 3L)))
       selector.collectPropagations() should ===(Map(nodes(0) → expected4))
 
-      val expected5 = DeltaPropagation(selfUniqueAddress, Map(
+      val expected5 = DeltaPropagation(selfUniqueAddress, false, Map(
         "A" → Delta(DataEnvelope(delta3), 3L, 3L)))
       selector.collectPropagations() should ===(Map(nodes(1) → expected5))
 
diff --git a/akka-distributed-data/src/test/scala/akka/cluster/ddata/WriteAggregatorSpec.scala b/akka-distributed-data/src/test/scala/akka/cluster/ddata/WriteAggregatorSpec.scala
index 2547a7cebe..a07209a18d 100644
--- a/akka-distributed-data/src/test/scala/akka/cluster/ddata/WriteAggregatorSpec.scala
+++ b/akka-distributed-data/src/test/scala/akka/cluster/ddata/WriteAggregatorSpec.scala
@@ -17,18 +17,24 @@ import akka.cluster.ddata.Replicator._
 import akka.remote.RARP
 
 import scala.concurrent.Future
+import akka.cluster.Cluster
 
 object WriteAggregatorSpec {
 
-  val key = GSetKey[String]("a")
+  val KeyA = GSetKey[String]("A")
+  val KeyB = ORSetKey[String]("B")
 
   def writeAggregatorProps(data: GSet[String], consistency: Replicator.WriteConsistency,
                            probes: Map[Address, ActorRef], nodes: Set[Address], unreachable: Set[Address], replyTo: ActorRef, durable: Boolean): Props =
-    Props(new TestWriteAggregator(data, consistency, probes, nodes, unreachable, replyTo, durable))
+    Props(new TestWriteAggregator(KeyA, data, None, consistency, probes, nodes, unreachable, replyTo, durable))
 
-  class TestWriteAggregator(data: GSet[String], consistency: Replicator.WriteConsistency,
+  def writeAggregatorPropsWithDelta(data: ORSet[String], delta: Delta, consistency: Replicator.WriteConsistency,
+                                    probes: Map[Address, ActorRef], nodes: Set[Address], unreachable: Set[Address], replyTo: ActorRef, durable: Boolean): Props =
+    Props(new TestWriteAggregator(KeyB, data, Some(delta), consistency, probes, nodes, unreachable, replyTo, durable))
+
+  class TestWriteAggregator(key: Key.KeyR, data: ReplicatedData, delta: Option[Delta], consistency: Replicator.WriteConsistency,
                             probes: Map[Address, ActorRef], nodes: Set[Address], unreachable: Set[Address], replyTo: ActorRef, durable: Boolean)
-    extends WriteAggregator(key, DataEnvelope(data), consistency, None, nodes, unreachable, replyTo, durable) {
+    extends WriteAggregator(key, DataEnvelope(data), delta, consistency, None, nodes, unreachable, replyTo, durable) {
 
     override def replica(address: Address): ActorSelection =
       context.actorSelection(probes(address).path)
@@ -48,6 +54,8 @@ object WriteAggregatorSpec {
         replicator.foreach(_ ! WriteAck)
       case WriteNack ⇒
         replicator.foreach(_ ! WriteNack)
+      case DeltaNack ⇒
+        replicator.foreach(_ ! DeltaNack)
       case msg ⇒
         replicator = Some(sender())
         replica ! msg
@@ -89,6 +97,7 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
   val timeout = 3.seconds.dilated
   val writeThree = WriteTo(3, timeout)
   val writeMajority = WriteMajority(timeout)
+  val writeAll = WriteAll(timeout)
 
   def probes(probe: ActorRef): Map[Address, ActorRef] =
     nodes.toSeq.map(_ → system.actorOf(WriteAggregatorSpec.writeAckAdapterProps(probe))).toMap
@@ -111,7 +120,7 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
       probe.lastSender ! WriteAck
       probe.expectMsgType[Write]
       probe.lastSender ! WriteAck
-      expectMsg(UpdateSuccess(WriteAggregatorSpec.key, None))
+      expectMsg(UpdateSuccess(WriteAggregatorSpec.KeyA, None))
       watch(aggr)
       expectTerminated(aggr)
     }
@@ -132,15 +141,14 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
       Future.sequence {
         Seq(
           Future { testProbes(nodeC).expectNoMsg(t) },
-          Future { testProbes(nodeD).expectNoMsg(t) }
-        )
+          Future { testProbes(nodeD).expectNoMsg(t) })
       }.futureValue
       testProbes(nodeC).expectMsgType[Write]
       testProbes(nodeC).lastSender ! WriteAck
       testProbes(nodeD).expectMsgType[Write]
       testProbes(nodeD).lastSender ! WriteAck
 
-      expectMsg(UpdateSuccess(WriteAggregatorSpec.key, None))
+      expectMsg(UpdateSuccess(WriteAggregatorSpec.KeyA, None))
       watch(aggr)
       expectTerminated(aggr)
     }
@@ -158,7 +166,7 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
       // no reply
       probe.expectMsgType[Write]
       // no reply
-      expectMsg(UpdateTimeout(WriteAggregatorSpec.key, None))
+      expectMsg(UpdateTimeout(WriteAggregatorSpec.KeyA, None))
       watch(aggr)
       expectTerminated(aggr)
     }
@@ -181,6 +189,82 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
     }
   }
 
+  "WriteAggregator with delta" must {
+    implicit val cluster = Cluster(system)
+    val fullState1 = ORSet.empty[String] + "a" + "b"
+    val fullState2 = fullState1.resetDelta + "c"
+    val delta = Delta(DataEnvelope(fullState2.delta.get), 2L, 2L)
+
+    "send deltas first" in {
+      val probe = TestProbe()
+      val aggr = system.actorOf(WriteAggregatorSpec.writeAggregatorPropsWithDelta(
+        fullState2, delta, writeMajority, probes(probe.ref), nodes, Set.empty, testActor, durable = false))
+
+      probe.expectMsgType[DeltaPropagation]
+      probe.lastSender ! WriteAck
+      probe.expectMsgType[DeltaPropagation]
+      probe.lastSender ! WriteAck
+      expectMsg(UpdateSuccess(WriteAggregatorSpec.KeyB, None))
+      watch(aggr)
+      expectTerminated(aggr)
+    }
+
+    "retry with full state when no immediate reply or nack" in {
+      val testProbes = probes()
+      val testProbeRefs = testProbes.map { case (a, tm) ⇒ a → tm.writeAckAdapter }
+      val aggr = system.actorOf(WriteAggregatorSpec.writeAggregatorPropsWithDelta(
+        fullState2, delta, writeAll, testProbeRefs, nodes, Set.empty, testActor, durable = false))
+
+      testProbes(nodeA).expectMsgType[DeltaPropagation]
+      // no reply
+      testProbes(nodeB).expectMsgType[DeltaPropagation]
+      testProbes(nodeB).lastSender ! WriteAck
+      testProbes(nodeC).expectMsgType[DeltaPropagation]
+      testProbes(nodeC).lastSender ! WriteAck
+      testProbes(nodeD).expectMsgType[DeltaPropagation]
+      testProbes(nodeD).lastSender ! DeltaNack
+
+      // here is the second round
+      testProbes(nodeA).expectMsgType[Write]
+      testProbes(nodeA).lastSender ! WriteAck
+      testProbes(nodeD).expectMsgType[Write]
+      testProbes(nodeD).lastSender ! WriteAck
+      testProbes(nodeB).expectNoMsg(100.millis)
+      testProbes(nodeC).expectNoMsg(100.millis)
+
+      expectMsg(UpdateSuccess(WriteAggregatorSpec.KeyB, None))
+      watch(aggr)
+      expectTerminated(aggr)
+    }
+
+    "timeout when less than required acks" in {
+      val probe = TestProbe()
+      val aggr = system.actorOf(WriteAggregatorSpec.writeAggregatorPropsWithDelta(
+        fullState2, delta, writeAll, probes(probe.ref), nodes, Set.empty, testActor, durable = false))
+
+      probe.expectMsgType[DeltaPropagation]
+      // no reply
+      probe.expectMsgType[DeltaPropagation]
+      probe.lastSender ! WriteAck
+      probe.expectMsgType[DeltaPropagation]
+      // nack
+      probe.lastSender ! DeltaNack
+      probe.expectMsgType[DeltaPropagation]
+      // no reply
+
+      // only 1 ack so we expect 3 full state Write
+      probe.expectMsgType[Write]
+      probe.lastSender ! WriteAck
+      probe.expectMsgType[Write]
+      probe.expectMsgType[Write]
+
+      // still not enough acks
+      expectMsg(UpdateTimeout(WriteAggregatorSpec.KeyB, None))
+      watch(aggr)
+      expectTerminated(aggr)
+    }
+  }
+
   "Durable WriteAggregator" must {
     "not reply before local confirmation" in {
       val probe = TestProbe()
@@ -194,9 +278,9 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
       expectNoMsg(200.millis)
 
       // the local write
-      aggr ! UpdateSuccess(WriteAggregatorSpec.key, None)
+      aggr ! UpdateSuccess(WriteAggregatorSpec.KeyA, None)
 
-      expectMsg(UpdateSuccess(WriteAggregatorSpec.key, None))
+      expectMsg(UpdateSuccess(WriteAggregatorSpec.KeyA, None))
       watch(aggr)
       expectTerminated(aggr)
     }
@@ -206,7 +290,7 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
       val aggr = system.actorOf(WriteAggregatorSpec.writeAggregatorProps(
         data, writeThree, probes(probe.ref), nodes, Set.empty, testActor, durable = true))
 
-      aggr ! UpdateSuccess(WriteAggregatorSpec.key, None) // the local write
+      aggr ! UpdateSuccess(WriteAggregatorSpec.KeyA, None) // the local write
       probe.expectMsgType[Write]
       probe.lastSender ! WriteAck
       probe.expectMsgType[Write]
@@ -214,7 +298,7 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
       probe.expectMsgType[Write]
       probe.lastSender ! WriteAck
 
-      expectMsg(UpdateSuccess(WriteAggregatorSpec.key, None))
+      expectMsg(UpdateSuccess(WriteAggregatorSpec.KeyA, None))
       watch(aggr)
       expectTerminated(aggr)
     }
@@ -226,7 +310,7 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
 
       probe.expectMsgType[Write]
       probe.lastSender ! WriteNack
-      aggr ! UpdateSuccess(WriteAggregatorSpec.key, None) // the local write
+      aggr ! UpdateSuccess(WriteAggregatorSpec.KeyA, None) // the local write
       probe.expectMsgType[Write]
       probe.lastSender ! WriteAck
       probe.expectMsgType[Write]
@@ -234,7 +318,7 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
       probe.expectMsgType[Write]
       probe.lastSender ! WriteNack
 
-      expectMsg(StoreFailure(WriteAggregatorSpec.key, None))
+      expectMsg(StoreFailure(WriteAggregatorSpec.KeyA, None))
       watch(aggr)
       expectTerminated(aggr)
     }
@@ -253,7 +337,7 @@ class WriteAggregatorSpec extends AkkaSpec(s"""
       probe.expectMsgType[Write]
       probe.lastSender ! WriteNack
 
-      expectMsg(UpdateTimeout(WriteAggregatorSpec.key, None))
+      expectMsg(UpdateTimeout(WriteAggregatorSpec.KeyA, None))
       watch(aggr)
       expectTerminated(aggr)
     }
diff --git a/akka-distributed-data/src/test/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializerSpec.scala b/akka-distributed-data/src/test/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializerSpec.scala
index d920da41af..ee0ee0fb59 100644
--- a/akka-distributed-data/src/test/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializerSpec.scala
+++ b/akka-distributed-data/src/test/scala/akka/cluster/ddata/protobuf/ReplicatorMessageSerializerSpec.scala
@@ -78,6 +78,7 @@ class ReplicatorMessageSerializerSpec extends TestKit(ActorSystem(
       checkSerialization(Write("A", DataEnvelope(data1)))
       checkSerialization(WriteAck)
       checkSerialization(WriteNack)
+      checkSerialization(DeltaNack)
       checkSerialization(Read("A"))
       checkSerialization(ReadResult(Some(DataEnvelope(data1))))
       checkSerialization(ReadResult(None))
@@ -87,7 +88,7 @@ class ReplicatorMessageSerializerSpec extends TestKit(ActorSystem(
       checkSerialization(Gossip(Map(
         "A" → DataEnvelope(data1),
         "B" → DataEnvelope(GSet() + "b" + "c")), sendBack = true))
-      checkSerialization(DeltaPropagation(address1, Map(
+      checkSerialization(DeltaPropagation(address1, reply = true, Map(
         "A" → Delta(DataEnvelope(delta1), 1L, 1L),
         "B" → Delta(DataEnvelope(delta2), 3L, 5L))))
       checkSerialization(new DurableDataEnvelope(data1))