Unwrap tagged replicated events before publishing to other replicas (#30136)

akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/ShardingDirectReplication.scala

@@ -76,7 +76,9 @@ private[akka] object ShardingDirectReplication {
               replicaShardingProxies.foreach {
                 case (replica, proxy) =>
                   val newId = replicationId.withReplica(replica)
-                  val envelopedEvent = ShardingEnvelope(newId.persistenceId.id, event)
+                  // receiving side is responsible for any tagging, so drop/unwrap any tags added by the local tagger
+                  val withoutTags = event.withoutTags
+                  val envelopedEvent = ShardingEnvelope(newId.persistenceId.id, withoutTags)
                   if (!selfReplica.contains(replica)) {
                     proxy.asInstanceOf[ActorRef[ShardingEnvelope[PublishedEvent]]] ! envelopedEvent
                   }

akka-persistence-typed/src/main/mima-filters/2.6.13.backwards.excludes/pr-30080-replicated-publishing-tagged-events.excludes

@@ -0,0 +1,2 @@
+# not for user extension
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.persistence.typed.PublishedEvent.withoutTags")

akka-persistence-typed/src/main/scala/akka/persistence/typed/PublishedEvent.scala

@@ -30,4 +30,10 @@ trait PublishedEvent {
   def event: Any
   def timestamp: Long
   def tags: Set[String]
+
+  /**
+   * If the published event is tagged, return a new published event with the payload unwrapped and the tags dropped,
+   * if it is not tagged return the published event as is.
+   */
+  def withoutTags: PublishedEvent
 }

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedBehaviorImpl.scala

@@ -377,5 +377,10 @@ private[akka] final case class PublishedEventImpl(
     case _                => payload
   }
 
+  override def withoutTags: PublishedEvent = payload match {
+    case Tagged(event, _) => copy(payload = event)
+    case _                => this
+  }
+
   override def getReplicatedMetaData: Optional[ReplicatedPublishedEventMetaData] = replicatedMetaData.asJava
 }