diff --git a/akka-actor-typed-tests/src/test/java/jdocs/akka/typed/LoggingDocExamples.java b/akka-actor-typed-tests/src/test/java/jdocs/akka/typed/LoggingDocExamples.java
index a7a59f743a..c31d4ac8b4 100644
--- a/akka-actor-typed-tests/src/test/java/jdocs/akka/typed/LoggingDocExamples.java
+++ b/akka-actor-typed-tests/src/test/java/jdocs/akka/typed/LoggingDocExamples.java
@@ -8,9 +8,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
-import akka.actor.typed.ActorRef;
-import akka.actor.typed.ActorSystem;
-import akka.actor.typed.Behavior;
+import akka.actor.typed.*;
 import akka.actor.typed.javadsl.AbstractBehavior;
 import akka.actor.typed.javadsl.ActorContext;
 import akka.actor.typed.javadsl.Behaviors;
@@ -19,7 +17,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 // #logMessages
-import akka.actor.typed.LogOptions;
 import org.slf4j.event.Level;
 
 // #logMessages
@@ -188,4 +185,12 @@ public interface LoggingDocExamples {
     // #test-logging-criteria
 
   }
+
+  static void tagsExample() {
+    ActorContext<Object> context = null;
+    Behavior<Object> myBehavior = Behaviors.empty();
+    // #tags
+    context.spawn(myBehavior, "MyActor", ActorTags.create("processing"));
+    // #tags
+  }
 }
diff --git a/akka-actor-typed-tests/src/test/scala/akka/actor/typed/scaladsl/ActorLoggingSpec.scala b/akka-actor-typed-tests/src/test/scala/akka/actor/typed/scaladsl/ActorLoggingSpec.scala
index fcc3a84ad4..d440882ab9 100644
--- a/akka-actor-typed-tests/src/test/scala/akka/actor/typed/scaladsl/ActorLoggingSpec.scala
+++ b/akka-actor-typed-tests/src/test/scala/akka/actor/typed/scaladsl/ActorLoggingSpec.scala
@@ -12,7 +12,9 @@ import akka.actor.testkit.typed.scaladsl.ActorTestKit
 import akka.actor.testkit.typed.scaladsl.LoggingTestKit
 import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
 import akka.actor.testkit.typed.scaladsl.LogCapturing
+import akka.actor.typed.ActorTags
 import akka.actor.typed.Behavior
+import akka.actor.typed.internal.ActorMdc
 import akka.actor.typed.scaladsl.adapter._
 import akka.event.DefaultLoggingFilter
 import akka.event.Logging.DefaultLogger
@@ -250,6 +252,26 @@ class ActorLoggingSpec extends ScalaTestWithActorTestKit("""
       }
     }
 
+    "pass tags from props to MDC" in {
+      val behavior = Behaviors.setup[String] { ctx =>
+        ctx.log.info("Starting up")
+
+        Behaviors.receiveMessage {
+          case msg =>
+            ctx.log.info("Got message {}", msg)
+            Behaviors.same
+        }
+      }
+      val actor =
+        LoggingTestKit.info("Starting up").withMdc(Map(ActorMdc.TagsKey -> "tag1,tag2")).intercept {
+          spawn(behavior, ActorTags("tag1", "tag2"))
+        }
+
+      LoggingTestKit.info("Got message").withMdc(Map(ActorMdc.TagsKey -> "tag1,tag2")).intercept {
+        actor ! "ping"
+      }
+    }
+
   }
 
   "SLF4J Settings" must {
diff --git a/akka-actor-typed-tests/src/test/scala/docs/akka/typed/LoggingDocExamples.scala b/akka-actor-typed-tests/src/test/scala/docs/akka/typed/LoggingDocExamples.scala
index aa6166f025..682dcff946 100644
--- a/akka-actor-typed-tests/src/test/scala/docs/akka/typed/LoggingDocExamples.scala
+++ b/akka-actor-typed-tests/src/test/scala/docs/akka/typed/LoggingDocExamples.scala
@@ -8,9 +8,9 @@ import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import scala.util.Failure
 import scala.util.Success
-
 import akka.actor.typed.ActorRef
 import akka.actor.typed.ActorSystem
+import akka.actor.typed.ActorTags
 import akka.actor.typed.Behavior
 import akka.actor.typed.scaladsl.Behaviors
 import org.slf4j.LoggerFactory
@@ -141,4 +141,14 @@ object LoggingDocExamples {
     //#test-logging-criteria
   }
 
+  def tagsExample(): Unit = {
+    Behaviors.setup[AnyRef] { context =>
+      val myBehavior = Behaviors.empty[AnyRef]
+      //#tags
+      context.spawn(myBehavior, "MyActor", ActorTags("processing"))
+      //#tags
+      Behaviors.stopped
+    }
+  }
+
 }
diff --git a/akka-actor-typed/src/main/scala/akka/actor/typed/Props.scala b/akka-actor-typed/src/main/scala/akka/actor/typed/Props.scala
index 6d41d781c8..dd1d213907 100644
--- a/akka-actor-typed/src/main/scala/akka/actor/typed/Props.scala
+++ b/akka-actor-typed/src/main/scala/akka/actor/typed/Props.scala
@@ -9,8 +9,10 @@ import akka.annotation.InternalApi
 
 import scala.annotation.tailrec
 import scala.reflect.ClassTag
-
 import akka.actor.typed.internal.PropsImpl._
+import akka.util.ccompat.JavaConverters._
+
+import scala.annotation.varargs
 
 object Props {
 
@@ -211,3 +213,57 @@ object MailboxSelector {
    */
   def fromConfig(path: String): MailboxSelector = MailboxFromConfigSelector(path)
 }
+
+/**
+ * Actor tags are used to logically group actors. The tags are included in logging as markers
+ * Especially useful for logging from functional style actors and since those may not have a clear logger class.
+ *
+ * Not for user extension.
+ */
+@DoNotInherit
+abstract class ActorTags extends Props {
+
+  /**
+   * Scala API: one or more tags defined for the actor
+   * @return
+   */
+  def tags: Set[String]
+
+  /**
+   * Java API: one or more tags defined for the actor
+   */
+  def getTags(): java.util.Set[String] = tags.asJava
+}
+
+object ActorTags {
+
+  /**
+   * Java API: create a tag props with one or more tags
+   */
+  @varargs
+  def create(tags: String*): ActorTags = apply(tags.toSet)
+
+  /**
+   * Java API: create a multi-tag props
+   *
+   * Set must not be empty.
+   */
+  def create(tags: java.util.Set[String]): ActorTags = ActorTagsImpl(tags.asScala.toSet)
+
+  /**
+   * Scala API: create a tag props with one or more tags
+   */
+  def apply(tag: String, additionalTags: String*): ActorTags = {
+    val tags =
+      if (additionalTags.isEmpty) Set(tag)
+      else Set(tag) ++ additionalTags
+    ActorTagsImpl(tags)
+  }
+
+  /**
+   * Scala API: create a multi-tag props.
+   *
+   * Set must not be empty.
+   */
+  def apply(tags: Set[String]): ActorTags = ActorTagsImpl(tags)
+}
diff --git a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/ActorContextImpl.scala b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/ActorContextImpl.scala
index 5ac9bbdf54..a74d414511 100644
--- a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/ActorContextImpl.scala
+++ b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/ActorContextImpl.scala
@@ -21,6 +21,34 @@ import com.github.ghik.silencer.silent
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
+/**
+ * INTERNAL API
+ */
+@InternalApi private[akka] object ActorContextImpl {
+
+  // single context for logging as there are a few things that are initialized
+  // together
+  final class LoggingContext(val logger: Logger, tags: Set[String]) {
+    // toggled once per message if logging is used to avoid having to
+    // touch the mdc thread local for cleanup in the end
+    var mdcUsed = false
+    val tagsString =
+      // "" means no tags
+      if (tags.isEmpty) ""
+      else
+        // mdc can only contain string values, and we don't want to render that string
+        // on each log entry or message, so do that up front here
+        tags.mkString(",")
+
+    def withLogger(logger: Logger): LoggingContext = {
+      val l = new LoggingContext(logger, tags)
+      l.mdcUsed = mdcUsed
+      l
+    }
+  }
+
+}
+
 /**
  * INTERNAL API
  */
@@ -29,9 +57,10 @@ import org.slf4j.LoggerFactory
     with javadsl.ActorContext[T]
     with scaladsl.ActorContext[T] {
 
+  import ActorContextImpl.LoggingContext
+
   // lazily initialized
-  private var logger: OptionVal[Logger] = OptionVal.None
-  private var mdcUsed: Boolean = false
+  private var _logging: OptionVal[LoggingContext] = OptionVal.None
 
   private var messageAdapterRef: OptionVal[ActorRef[Any]] = OptionVal.None
   private var _messageAdapters: List[(Class[_], Any => T)] = Nil
@@ -80,42 +109,47 @@ import org.slf4j.LoggerFactory
   override def getSystem: akka.actor.typed.ActorSystem[Void] =
     system.asInstanceOf[ActorSystem[Void]]
 
-  override def log: Logger = {
-    val l = logger match {
+  private def loggingContext(): LoggingContext = {
+    // lazy init of logging setup
+    _logging match {
       case OptionVal.Some(l) => l
       case OptionVal.None =>
         val logClass = LoggerClass.detectLoggerClassFromStack(classOf[Behavior[_]])
-        initLoggerWithName(logClass.getName)
+        val logger = LoggerFactory.getLogger(logClass.getName)
+        val l = new LoggingContext(logger, classicActorContext.props.deploy.tags)
+        _logging = OptionVal.Some(l)
+        l
     }
-    // avoid access to MDC ThreadLocal if not needed
-    mdcUsed = true
-    ActorMdc.setMdc(self.path.toString)
-    l
+  }
+
+  override def log: Logger = {
+    val logging = loggingContext()
+    // avoid access to MDC ThreadLocal if not needed, see details in LoggingContext
+    logging.mdcUsed = true
+    ActorMdc.setMdc(self.path.toString, logging.tagsString)
+    logging.logger
   }
 
   override def getLog: Logger = log
 
-  override def setLoggerName(name: String): Unit =
-    initLoggerWithName(name)
+  override def setLoggerName(name: String): Unit = {
+    _logging = OptionVal.Some(loggingContext().withLogger(LoggerFactory.getLogger(name)))
+  }
 
   override def setLoggerName(clazz: Class[_]): Unit =
     setLoggerName(clazz.getName)
 
   // MDC is cleared (if used) from aroundReceive in ActorAdapter after processing each message
   override private[akka] def clearMdc(): Unit = {
-    // avoid access to MDC ThreadLocal if not needed
-    if (mdcUsed) {
-      ActorMdc.clearMdc()
-      mdcUsed = false
+    // avoid access to MDC ThreadLocal if not needed, see details in LoggingContext
+    _logging match {
+      case OptionVal.Some(ctx) if ctx.mdcUsed =>
+        ActorMdc.clearMdc()
+        ctx.mdcUsed = false
+      case _ =>
     }
   }
 
-  private def initLoggerWithName(name: String): Logger = {
-    val l = LoggerFactory.getLogger(name)
-    logger = OptionVal.Some(l)
-    l
-  }
-
   override def setReceiveTimeout(d: java.time.Duration, msg: T): Unit =
     setReceiveTimeout(d.asScala, msg)
 
diff --git a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/ActorMdc.scala b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/ActorMdc.scala
index e1c4c6c98a..d02653566f 100644
--- a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/ActorMdc.scala
+++ b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/ActorMdc.scala
@@ -12,10 +12,16 @@ import org.slf4j.MDC
  */
 @InternalApi private[akka] object ActorMdc {
   val SourceKey = "akkaSource"
+  val TagsKey = "akkaTags"
 
-  def setMdc(source: String): Unit = {
-    val mdcAdpater = MDC.getMDCAdapter
-    mdcAdpater.put(SourceKey, source)
+  /**
+   * @param tags empty string for no tags, a single tag or a comma separated list of tags
+   */
+  def setMdc(source: String, tags: String): Unit = {
+    val mdcAdapter = MDC.getMDCAdapter
+    mdcAdapter.put(SourceKey, source)
+    if (tags.nonEmpty)
+      mdcAdapter.put(TagsKey, tags)
   }
 
   // MDC is cleared (if used) from aroundReceive in ActorAdapter after processing each message,
diff --git a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/PropsImpl.scala b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/PropsImpl.scala
index 6b4aa89577..55345c7038 100644
--- a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/PropsImpl.scala
+++ b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/PropsImpl.scala
@@ -4,6 +4,7 @@
 
 package akka.actor.typed.internal
 
+import akka.actor.typed.ActorTags
 import akka.actor.typed.{ DispatcherSelector, MailboxSelector, Props }
 import akka.annotation.InternalApi
 
@@ -54,4 +55,14 @@ import akka.annotation.InternalApi
     def withNext(next: Props): Props = copy(next = next)
   }
 
+  final case class ActorTagsImpl(tags: Set[String], next: Props = Props.empty) extends ActorTags {
+    if (tags == null)
+      throw new IllegalArgumentException("Tags must not be null")
+    def withNext(next: Props): Props = copy(next = next)
+  }
+
+  object ActorTagsImpl {
+    val empty = ActorTagsImpl(Set.empty)
+  }
+
 }
diff --git a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/adapter/PropsAdapter.scala b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/adapter/PropsAdapter.scala
index c4d4b94166..dd5da04a89 100644
--- a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/adapter/PropsAdapter.scala
+++ b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/adapter/PropsAdapter.scala
@@ -5,6 +5,7 @@
 package akka.actor.typed.internal.adapter
 
 import akka.actor.Deploy
+import akka.actor.typed.ActorTags
 import akka.actor.typed.Behavior
 import akka.actor.typed.DispatcherSelector
 import akka.actor.typed.MailboxSelector
@@ -38,7 +39,11 @@ import akka.dispatch.Mailboxes
         dispatcherProps.withMailbox(path)
     }
 
-    mailboxProps.withDeploy(Deploy.local) // disallow remote deployment for typed actors
+    val localDeploy = mailboxProps.withDeploy(Deploy.local) // disallow remote deployment for typed actors
+
+    val tags = deploy.firstOrElse[ActorTags](ActorTagsImpl.empty).tags
+    if (tags.isEmpty) localDeploy
+    else localDeploy.withActorTags(tags)
   }
 
 }
diff --git a/akka-actor/src/main/scala/akka/actor/Deployer.scala b/akka-actor/src/main/scala/akka/actor/Deployer.scala
index e15aa35d96..96c445988c 100644
--- a/akka-actor/src/main/scala/akka/actor/Deployer.scala
+++ b/akka-actor/src/main/scala/akka/actor/Deployer.scala
@@ -23,6 +23,19 @@ object Deploy {
    * INTERNAL API
    */
   @InternalApi private[akka] final val DispatcherSameAsParent = ".."
+
+  def apply(
+      path: String = "",
+      config: Config = ConfigFactory.empty,
+      routerConfig: RouterConfig = NoRouter,
+      scope: Scope = NoScopeGiven,
+      dispatcher: String = Deploy.NoDispatcherGiven,
+      mailbox: String = Deploy.NoMailboxGiven): Deploy =
+    new Deploy(path, config, routerConfig, scope, dispatcher, mailbox, Set.empty)
+
+  // for bincomp, pre 2.6 was case class
+  def unapply(deploy: Deploy): Option[(String, Config, RouterConfig, Scope, String, String)] =
+    Some((deploy.path, deploy.config, deploy.routerConfig, deploy.scope, deploy.dispatcher, deploy.mailbox))
 }
 
 /**
@@ -40,14 +53,27 @@ object Deploy {
  * val remoteProps = someProps.withDeploy(Deploy(scope = RemoteScope("someOtherNodeName")))
  * }}}
  */
-@SerialVersionUID(2L)
-final case class Deploy(
-    path: String = "",
-    config: Config = ConfigFactory.empty,
-    routerConfig: RouterConfig = NoRouter,
-    scope: Scope = NoScopeGiven,
-    dispatcher: String = Deploy.NoDispatcherGiven,
-    mailbox: String = Deploy.NoMailboxGiven) {
+@SerialVersionUID(3L)
+final class Deploy(
+    val path: String = "",
+    val config: Config = ConfigFactory.empty,
+    val routerConfig: RouterConfig = NoRouter,
+    val scope: Scope = NoScopeGiven,
+    val dispatcher: String = Deploy.NoDispatcherGiven,
+    val mailbox: String = Deploy.NoMailboxGiven,
+    val tags: Set[String] = Set.empty)
+    extends Serializable
+    with Product
+    with Equals {
+
+  // for bincomp, pre 2.6 did not have tags
+  def this(
+      path: String,
+      config: Config,
+      routerConfig: RouterConfig,
+      scope: Scope,
+      dispatcher: String,
+      mailbox: String) = this(path, config, routerConfig, scope, dispatcher, mailbox, Set.empty)
 
   /**
    * Java API to create a Deploy with the given RouterConfig
@@ -78,6 +104,52 @@ final case class Deploy(
       if (dispatcher == Deploy.NoDispatcherGiven) other.dispatcher else dispatcher,
       if (mailbox == Deploy.NoMailboxGiven) other.mailbox else mailbox)
   }
+
+  def withTags(tags: Set[String]): Deploy =
+    new Deploy(path, config, routerConfig, scope, dispatcher, mailbox, tags)
+
+  // below are for bincomp, pre 2.6 was case class
+  def copy(
+      path: String = path,
+      config: Config = config,
+      routerConfig: RouterConfig = routerConfig,
+      scope: Scope = scope,
+      dispatcher: String = dispatcher,
+      mailbox: String = mailbox): Deploy =
+    new Deploy(path, config, routerConfig, scope, dispatcher, mailbox, tags)
+
+  override def productElement(n: Int): Any = n match {
+    case 1 => path
+    case 2 => config
+    case 3 => routerConfig
+    case 4 => scope
+    case 5 => dispatcher
+    case 6 => mailbox
+    case 7 => tags
+  }
+
+  override def productArity: Int = 7
+
+  override def canEqual(that: Any): Boolean = that.isInstanceOf[Deploy]
+
+  override def equals(other: Any): Boolean = other match {
+    case that: Deploy =>
+      path == that.path &&
+      config == that.config &&
+      routerConfig == that.routerConfig &&
+      scope == that.scope &&
+      dispatcher == that.dispatcher &&
+      mailbox == that.mailbox &&
+      tags == that.tags
+    case _ => false
+  }
+
+  override def hashCode(): Int = {
+    val state = Seq[AnyRef](path, config, routerConfig, scope, dispatcher, mailbox, tags)
+    state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b)
+  }
+
+  override def toString = s"Deploy($path, $config, $routerConfig, $scope, $dispatcher, $mailbox, $tags)"
 }
 
 /**
diff --git a/akka-actor/src/main/scala/akka/actor/Props.scala b/akka-actor/src/main/scala/akka/actor/Props.scala
index ba19214c2a..e1f1f87545 100644
--- a/akka-actor/src/main/scala/akka/actor/Props.scala
+++ b/akka-actor/src/main/scala/akka/actor/Props.scala
@@ -8,6 +8,7 @@ import akka.actor.Deploy.{ NoDispatcherGiven, NoMailboxGiven }
 import akka.dispatch._
 import akka.routing._
 
+import scala.annotation.varargs
 import scala.collection.immutable
 import scala.reflect.ClassTag
 
@@ -192,6 +193,19 @@ final case class Props(deploy: Deploy, clazz: Class[_], args: immutable.Seq[Any]
    */
   def withDeploy(d: Deploy): Props = copy(deploy = d.withFallback(deploy))
 
+  /**
+   * Returns a new Props with the specified set of tags.
+   */
+  @varargs
+  def withActorTags(tags: String*): Props =
+    withActorTags(tags.toSet)
+
+  /**
+   * Scala API: Returns a new Props with the specified set of tags.
+   */
+  def withActorTags(tags: Set[String]): Props =
+    copy(deploy = deploy.withTags(tags))
+
   /**
    * Obtain an upper-bound approximation of the actor class which is going to
    * be created by these Props. In other words, the actor factory method will
diff --git a/akka-docs/src/main/paradox/typed/logging.md b/akka-docs/src/main/paradox/typed/logging.md
index c1c624b824..ea8f081cdb 100644
--- a/akka-docs/src/main/paradox/typed/logging.md
+++ b/akka-docs/src/main/paradox/typed/logging.md
@@ -126,10 +126,23 @@ Java
 :  @@snip [LoggingDocExamples.java](/akka-actor-typed-tests/src/test/java/jdocs/akka/typed/LoggingDocExamples.java) { #logMessages }
 
 
-## Behaviors.withMdc
+## MDC
 
-To include [MDC](http://logback.qos.ch/manual/mdc.html) attributes in logging events from an actor
-you can decorate a `Behavior` with `Behaviors.withMdc` or use the `org.slf4j.MDC` API directly.
+[MDC](http://logback.qos.ch/manual/mdc.html) allows for adding additional context dependent attributes to log entries.
+Out of the box Akka will place the path of the actor in the the MDC attribute `akkaSource`.
+
+One or more tags can also be added to the MDC using the `ActorTags` props. The tags will be rendered as a comma separated
+list and be put in the MDC attribute `akkaTags`. This can be used to categorize log entries from a set of different actors
+to allow easier filtering of logs:
+
+Scala
+:  @@snip [LoggingDocExamples.scala](/akka-actor-typed-tests/src/test/scala/docs/akka/typed/LoggingDocExamples.scala) { #tags }
+
+Java
+:  @@snip [LoggingDocExamples.java](/akka-actor-typed-tests/src/test/java/jdocs/akka/typed/LoggingDocExamples.java) { #tags }
+
+In addition to these two built in MDC attributes you can also decorate a `Behavior` with `Behaviors.withMdc` or 
+use the `org.slf4j.MDC` API directly.
 
 The `Behaviors.withMdc` decorator has two parts. A static `Map` of MDC attributes that are not changed,
 and a dynamic `Map` that can be constructed for each message.
diff --git a/akka-remote/src/main/java/akka/remote/WireFormats.java b/akka-remote/src/main/java/akka/remote/WireFormats.java
index 2f18dc48bf..939e41f30e 100644
--- a/akka-remote/src/main/java/akka/remote/WireFormats.java
+++ b/akka-remote/src/main/java/akka/remote/WireFormats.java
@@ -7380,6 +7380,25 @@ public final class WireFormats {
      */
     akka.protobufv3.internal.ByteString
         getRouterConfigManifestBytes();
+
+    /**
+     * <code>repeated string tags = 12;</code>
+     */
+    java.util.List<java.lang.String>
+        getTagsList();
+    /**
+     * <code>repeated string tags = 12;</code>
+     */
+    int getTagsCount();
+    /**
+     * <code>repeated string tags = 12;</code>
+     */
+    java.lang.String getTags(int index);
+    /**
+     * <code>repeated string tags = 12;</code>
+     */
+    akka.protobufv3.internal.ByteString
+        getTagsBytes(int index);
   }
   /**
    * <pre>
@@ -7407,6 +7426,7 @@ public final class WireFormats {
       scopeManifest_ = "";
       configManifest_ = "";
       routerConfigManifest_ = "";
+      tags_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
     }
 
     @java.lang.Override
@@ -7500,6 +7520,15 @@ public final class WireFormats {
               routerConfigManifest_ = bs;
               break;
             }
+            case 98: {
+              akka.protobufv3.internal.ByteString bs = input.readBytes();
+              if (!((mutable_bitField0_ & 0x00000800) != 0)) {
+                tags_ = new akka.protobufv3.internal.LazyStringArrayList();
+                mutable_bitField0_ |= 0x00000800;
+              }
+              tags_.add(bs);
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -7515,6 +7544,9 @@ public final class WireFormats {
         throw new akka.protobufv3.internal.InvalidProtocolBufferException(
             e).setUnfinishedMessage(this);
       } finally {
+        if (((mutable_bitField0_ & 0x00000800) != 0)) {
+          tags_ = tags_.getUnmodifiableView();
+        }
         this.unknownFields = unknownFields.build();
         makeExtensionsImmutable();
       }
@@ -7843,6 +7875,35 @@ public final class WireFormats {
       }
     }
 
+    public static final int TAGS_FIELD_NUMBER = 12;
+    private akka.protobufv3.internal.LazyStringList tags_;
+    /**
+     * <code>repeated string tags = 12;</code>
+     */
+    public akka.protobufv3.internal.ProtocolStringList
+        getTagsList() {
+      return tags_;
+    }
+    /**
+     * <code>repeated string tags = 12;</code>
+     */
+    public int getTagsCount() {
+      return tags_.size();
+    }
+    /**
+     * <code>repeated string tags = 12;</code>
+     */
+    public java.lang.String getTags(int index) {
+      return tags_.get(index);
+    }
+    /**
+     * <code>repeated string tags = 12;</code>
+     */
+    public akka.protobufv3.internal.ByteString
+        getTagsBytes(int index) {
+      return tags_.getByteString(index);
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -7894,6 +7955,9 @@ public final class WireFormats {
       if (((bitField0_ & 0x00000400) != 0)) {
         akka.protobufv3.internal.GeneratedMessageV3.writeString(output, 11, routerConfigManifest_);
       }
+      for (int i = 0; i < tags_.size(); i++) {
+        akka.protobufv3.internal.GeneratedMessageV3.writeString(output, 12, tags_.getRaw(i));
+      }
       unknownFields.writeTo(output);
     }
 
@@ -7942,6 +8006,14 @@ public final class WireFormats {
       if (((bitField0_ & 0x00000400) != 0)) {
         size += akka.protobufv3.internal.GeneratedMessageV3.computeStringSize(11, routerConfigManifest_);
       }
+      {
+        int dataSize = 0;
+        for (int i = 0; i < tags_.size(); i++) {
+          dataSize += computeStringSizeNoTag(tags_.getRaw(i));
+        }
+        size += dataSize;
+        size += 1 * getTagsList().size();
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -8012,6 +8084,8 @@ public final class WireFormats {
         if (!getRouterConfigManifest()
             .equals(other.getRouterConfigManifest())) return false;
       }
+      if (!getTagsList()
+          .equals(other.getTagsList())) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -8067,6 +8141,10 @@ public final class WireFormats {
         hash = (37 * hash) + ROUTERCONFIGMANIFEST_FIELD_NUMBER;
         hash = (53 * hash) + getRouterConfigManifest().hashCode();
       }
+      if (getTagsCount() > 0) {
+        hash = (37 * hash) + TAGS_FIELD_NUMBER;
+        hash = (53 * hash) + getTagsList().hashCode();
+      }
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -8227,6 +8305,8 @@ public final class WireFormats {
         bitField0_ = (bitField0_ & ~0x00000200);
         routerConfigManifest_ = "";
         bitField0_ = (bitField0_ & ~0x00000400);
+        tags_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
+        bitField0_ = (bitField0_ & ~0x00000800);
         return this;
       }
 
@@ -8299,6 +8379,11 @@ public final class WireFormats {
           to_bitField0_ |= 0x00000400;
         }
         result.routerConfigManifest_ = routerConfigManifest_;
+        if (((bitField0_ & 0x00000800) != 0)) {
+          tags_ = tags_.getUnmodifiableView();
+          bitField0_ = (bitField0_ & ~0x00000800);
+        }
+        result.tags_ = tags_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -8391,6 +8476,16 @@ public final class WireFormats {
           routerConfigManifest_ = other.routerConfigManifest_;
           onChanged();
         }
+        if (!other.tags_.isEmpty()) {
+          if (tags_.isEmpty()) {
+            tags_ = other.tags_;
+            bitField0_ = (bitField0_ & ~0x00000800);
+          } else {
+            ensureTagsIsMutable();
+            tags_.addAll(other.tags_);
+          }
+          onChanged();
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -9024,6 +9119,99 @@ public final class WireFormats {
         onChanged();
         return this;
       }
+
+      private akka.protobufv3.internal.LazyStringList tags_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
+      private void ensureTagsIsMutable() {
+        if (!((bitField0_ & 0x00000800) != 0)) {
+          tags_ = new akka.protobufv3.internal.LazyStringArrayList(tags_);
+          bitField0_ |= 0x00000800;
+         }
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public akka.protobufv3.internal.ProtocolStringList
+          getTagsList() {
+        return tags_.getUnmodifiableView();
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public int getTagsCount() {
+        return tags_.size();
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public java.lang.String getTags(int index) {
+        return tags_.get(index);
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public akka.protobufv3.internal.ByteString
+          getTagsBytes(int index) {
+        return tags_.getByteString(index);
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public Builder setTags(
+          int index, java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  ensureTagsIsMutable();
+        tags_.set(index, value);
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public Builder addTags(
+          java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  ensureTagsIsMutable();
+        tags_.add(value);
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public Builder addAllTags(
+          java.lang.Iterable<java.lang.String> values) {
+        ensureTagsIsMutable();
+        akka.protobufv3.internal.AbstractMessageLite.Builder.addAll(
+            values, tags_);
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public Builder clearTags() {
+        tags_ = akka.protobufv3.internal.LazyStringArrayList.EMPTY;
+        bitField0_ = (bitField0_ & ~0x00000800);
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>repeated string tags = 12;</code>
+       */
+      public Builder addTagsBytes(
+          akka.protobufv3.internal.ByteString value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  ensureTagsIsMutable();
+        tags_.add(value);
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final akka.protobufv3.internal.UnknownFieldSet unknownFields) {
@@ -19678,48 +19866,48 @@ public final class WireFormats {
       "\002(\0132\r.ActorRefData\"\204\001\n\tPropsData\022\033\n\006depl" +
       "oy\030\002 \002(\0132\013.DeployData\022\r\n\005clazz\030\003 \002(\t\022\014\n\004" +
       "args\030\004 \003(\014\022\021\n\tmanifests\030\005 \003(\t\022\025\n\rseriali" +
-      "zerIds\030\006 \003(\005\022\023\n\013hasManifest\030\007 \003(\010\"\211\002\n\nDe" +
+      "zerIds\030\006 \003(\005\022\023\n\013hasManifest\030\007 \003(\010\"\227\002\n\nDe" +
       "ployData\022\014\n\004path\030\001 \002(\t\022\016\n\006config\030\002 \001(\014\022\024" +
       "\n\014routerConfig\030\003 \001(\014\022\r\n\005scope\030\004 \001(\014\022\022\n\nd" +
       "ispatcher\030\005 \001(\t\022\031\n\021scopeSerializerId\030\006 \001" +
       "(\005\022\025\n\rscopeManifest\030\007 \001(\t\022\032\n\022configSeria" +
       "lizerId\030\010 \001(\005\022\026\n\016configManifest\030\t \001(\t\022 \n" +
       "\030routerConfigSerializerId\030\n \001(\005\022\034\n\024route" +
-      "rConfigManifest\030\013 \001(\t\"P\n\023AkkaProtocolMes" +
-      "sage\022\017\n\007payload\030\001 \001(\014\022(\n\013instruction\030\002 \001" +
-      "(\0132\023.AkkaControlMessage\"b\n\022AkkaControlMe" +
-      "ssage\022!\n\013commandType\030\001 \002(\0162\014.CommandType" +
-      "\022)\n\rhandshakeInfo\030\002 \001(\0132\022.AkkaHandshakeI" +
-      "nfo\"N\n\021AkkaHandshakeInfo\022\034\n\006origin\030\001 \002(\013" +
-      "2\014.AddressData\022\013\n\003uid\030\002 \002(\006\022\016\n\006cookie\030\003 " +
-      "\001(\t\"8\n\016FiniteDuration\022\r\n\005value\030\001 \002(\003\022\027\n\004" +
-      "unit\030\002 \002(\0162\t.TimeUnit\")\n\013RemoteScope\022\032\n\004" +
-      "node\030\001 \002(\0132\014.AddressData\"\261\001\n\016DefaultResi" +
-      "zer\022\022\n\nlowerBound\030\001 \002(\r\022\022\n\nupperBound\030\002 " +
-      "\002(\r\022\031\n\021pressureThreshold\030\003 \002(\r\022\022\n\nrampup" +
-      "Rate\030\004 \002(\001\022\030\n\020backoffThreshold\030\005 \002(\001\022\023\n\013" +
-      "backoffRate\030\006 \002(\001\022\031\n\021messagesPerResize\030\007" +
-      " \002(\r\"A\n\nFromConfig\022\031\n\007resizer\030\001 \001(\0132\010.Pa" +
-      "yload\022\030\n\020routerDispatcher\030\002 \001(\t\"{\n\022Gener" +
-      "icRoutingPool\022\025\n\rnrOfInstances\030\001 \002(\r\022\030\n\020" +
-      "routerDispatcher\030\002 \001(\t\022\031\n\021usePoolDispatc" +
-      "her\030\003 \002(\010\022\031\n\007resizer\030\004 \001(\0132\010.Payload\"Z\n\021" +
-      "ScatterGatherPool\022$\n\007generic\030\001 \002(\0132\023.Gen" +
-      "ericRoutingPool\022\037\n\006within\030\002 \002(\0132\017.Finite" +
-      "Duration\"|\n\020TailChoppingPool\022$\n\007generic\030" +
-      "\001 \002(\0132\023.GenericRoutingPool\022\037\n\006within\030\002 \002" +
-      "(\0132\017.FiniteDuration\022!\n\010interval\030\003 \002(\0132\017." +
-      "FiniteDuration\"O\n\013AddressData\022\016\n\006system\030" +
-      "\001 \002(\t\022\020\n\010hostname\030\002 \002(\t\022\014\n\004port\030\003 \002(\r\022\020\n" +
-      "\010protocol\030\004 \001(\t\"J\n\022RemoteRouterConfig\022\027\n" +
-      "\005local\030\001 \002(\0132\010.Payload\022\033\n\005nodes\030\002 \003(\0132\014." +
-      "AddressData*{\n\013CommandType\022\r\n\tASSOCIATE\020" +
-      "\001\022\020\n\014DISASSOCIATE\020\002\022\r\n\tHEARTBEAT\020\003\022\036\n\032DI" +
-      "SASSOCIATE_SHUTTING_DOWN\020\004\022\034\n\030DISASSOCIA" +
-      "TE_QUARANTINED\020\005*n\n\010TimeUnit\022\017\n\013NANOSECO" +
-      "NDS\020\001\022\020\n\014MICROSECONDS\020\002\022\020\n\014MILLISECONDS\020" +
-      "\003\022\013\n\007SECONDS\020\004\022\013\n\007MINUTES\020\005\022\t\n\005HOURS\020\006\022\010" +
-      "\n\004DAYS\020\007B\017\n\013akka.remoteH\001"
+      "rConfigManifest\030\013 \001(\t\022\014\n\004tags\030\014 \003(\t\"P\n\023A" +
+      "kkaProtocolMessage\022\017\n\007payload\030\001 \001(\014\022(\n\013i" +
+      "nstruction\030\002 \001(\0132\023.AkkaControlMessage\"b\n" +
+      "\022AkkaControlMessage\022!\n\013commandType\030\001 \002(\016" +
+      "2\014.CommandType\022)\n\rhandshakeInfo\030\002 \001(\0132\022." +
+      "AkkaHandshakeInfo\"N\n\021AkkaHandshakeInfo\022\034" +
+      "\n\006origin\030\001 \002(\0132\014.AddressData\022\013\n\003uid\030\002 \002(" +
+      "\006\022\016\n\006cookie\030\003 \001(\t\"8\n\016FiniteDuration\022\r\n\005v" +
+      "alue\030\001 \002(\003\022\027\n\004unit\030\002 \002(\0162\t.TimeUnit\")\n\013R" +
+      "emoteScope\022\032\n\004node\030\001 \002(\0132\014.AddressData\"\261" +
+      "\001\n\016DefaultResizer\022\022\n\nlowerBound\030\001 \002(\r\022\022\n" +
+      "\nupperBound\030\002 \002(\r\022\031\n\021pressureThreshold\030\003" +
+      " \002(\r\022\022\n\nrampupRate\030\004 \002(\001\022\030\n\020backoffThres" +
+      "hold\030\005 \002(\001\022\023\n\013backoffRate\030\006 \002(\001\022\031\n\021messa" +
+      "gesPerResize\030\007 \002(\r\"A\n\nFromConfig\022\031\n\007resi" +
+      "zer\030\001 \001(\0132\010.Payload\022\030\n\020routerDispatcher\030" +
+      "\002 \001(\t\"{\n\022GenericRoutingPool\022\025\n\rnrOfInsta" +
+      "nces\030\001 \002(\r\022\030\n\020routerDispatcher\030\002 \001(\t\022\031\n\021" +
+      "usePoolDispatcher\030\003 \002(\010\022\031\n\007resizer\030\004 \001(\013" +
+      "2\010.Payload\"Z\n\021ScatterGatherPool\022$\n\007gener" +
+      "ic\030\001 \002(\0132\023.GenericRoutingPool\022\037\n\006within\030" +
+      "\002 \002(\0132\017.FiniteDuration\"|\n\020TailChoppingPo" +
+      "ol\022$\n\007generic\030\001 \002(\0132\023.GenericRoutingPool" +
+      "\022\037\n\006within\030\002 \002(\0132\017.FiniteDuration\022!\n\010int" +
+      "erval\030\003 \002(\0132\017.FiniteDuration\"O\n\013AddressD" +
+      "ata\022\016\n\006system\030\001 \002(\t\022\020\n\010hostname\030\002 \002(\t\022\014\n" +
+      "\004port\030\003 \002(\r\022\020\n\010protocol\030\004 \001(\t\"J\n\022RemoteR" +
+      "outerConfig\022\027\n\005local\030\001 \002(\0132\010.Payload\022\033\n\005" +
+      "nodes\030\002 \003(\0132\014.AddressData*{\n\013CommandType" +
+      "\022\r\n\tASSOCIATE\020\001\022\020\n\014DISASSOCIATE\020\002\022\r\n\tHEA" +
+      "RTBEAT\020\003\022\036\n\032DISASSOCIATE_SHUTTING_DOWN\020\004" +
+      "\022\034\n\030DISASSOCIATE_QUARANTINED\020\005*n\n\010TimeUn" +
+      "it\022\017\n\013NANOSECONDS\020\001\022\020\n\014MICROSECONDS\020\002\022\020\n" +
+      "\014MILLISECONDS\020\003\022\013\n\007SECONDS\020\004\022\013\n\007MINUTES\020" +
+      "\005\022\t\n\005HOURS\020\006\022\010\n\004DAYS\020\007B\017\n\013akka.remoteH\001"
     };
     descriptor = akka.protobufv3.internal.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -19773,7 +19961,7 @@ public final class WireFormats {
     internal_static_DeployData_fieldAccessorTable = new
       akka.protobufv3.internal.GeneratedMessageV3.FieldAccessorTable(
         internal_static_DeployData_descriptor,
-        new java.lang.String[] { "Path", "Config", "RouterConfig", "Scope", "Dispatcher", "ScopeSerializerId", "ScopeManifest", "ConfigSerializerId", "ConfigManifest", "RouterConfigSerializerId", "RouterConfigManifest", });
+        new java.lang.String[] { "Path", "Config", "RouterConfig", "Scope", "Dispatcher", "ScopeSerializerId", "ScopeManifest", "ConfigSerializerId", "ConfigManifest", "RouterConfigSerializerId", "RouterConfigManifest", "Tags", });
     internal_static_AkkaProtocolMessage_descriptor =
       getDescriptor().getMessageTypes().get(8);
     internal_static_AkkaProtocolMessage_fieldAccessorTable = new
diff --git a/akka-remote/src/main/protobuf/WireFormats.proto b/akka-remote/src/main/protobuf/WireFormats.proto
index beb73cf49b..8bc78e2dcb 100644
--- a/akka-remote/src/main/protobuf/WireFormats.proto
+++ b/akka-remote/src/main/protobuf/WireFormats.proto
@@ -96,6 +96,7 @@ message DeployData {
   optional string configManifest = 9;
   optional int32 routerConfigSerializerId = 10;
   optional string routerConfigManifest = 11;
+  repeated string tags = 12;
 }
 
 
diff --git a/akka-remote/src/main/scala/akka/remote/serialization/DaemonMsgCreateSerializer.scala b/akka-remote/src/main/scala/akka/remote/serialization/DaemonMsgCreateSerializer.scala
index 0c27ccc5e0..3d25d3b39c 100644
--- a/akka-remote/src/main/scala/akka/remote/serialization/DaemonMsgCreateSerializer.scala
+++ b/akka-remote/src/main/scala/akka/remote/serialization/DaemonMsgCreateSerializer.scala
@@ -17,6 +17,7 @@ import akka.util.ccompat._
 
 import scala.reflect.ClassTag
 import util.{ Failure, Success }
+import akka.util.ccompat.JavaConverters._
 
 /**
  * Serializes Akka's internal DaemonMsgCreate using protobuf
@@ -66,6 +67,9 @@ private[akka] final class DaemonMsgCreateSerializer(val system: ExtendedActorSys
         if (d.dispatcher != NoDispatcherGiven) {
           builder.setDispatcher(d.dispatcher)
         }
+        if (d.tags.nonEmpty) {
+          builder.addAllTags(d.tags.asJava)
+        }
         builder.build
       }
 
@@ -149,7 +153,13 @@ private[akka] final class DaemonMsgCreateSerializer(val system: ExtendedActorSys
       val dispatcher =
         if (protoDeploy.hasDispatcher) protoDeploy.getDispatcher
         else NoDispatcherGiven
-      Deploy(protoDeploy.getPath, config, routerConfig, scope, dispatcher)
+
+      val tags: Set[String] =
+        if (protoDeploy.getTagsCount == 0) Set.empty
+        else protoDeploy.getTagsList.iterator().asScala.toSet
+      val deploy = Deploy(protoDeploy.getPath, config, routerConfig, scope, dispatcher)
+      if (tags.isEmpty) deploy
+      else deploy.withTags(tags)
     }
 
     def props = {
diff --git a/akka-remote/src/test/scala/akka/remote/serialization/MessageContainerSerializerSpec.scala b/akka-remote/src/test/scala/akka/remote/serialization/MessageContainerSerializerSpec.scala
index 04bc4d6c0a..fa45662e34 100644
--- a/akka-remote/src/test/scala/akka/remote/serialization/MessageContainerSerializerSpec.scala
+++ b/akka-remote/src/test/scala/akka/remote/serialization/MessageContainerSerializerSpec.scala
@@ -4,12 +4,14 @@
 
 package akka.remote.serialization
 
-import akka.serialization.SerializationExtension
-import akka.testkit.AkkaSpec
 import akka.actor.ActorSelectionMessage
 import akka.actor.SelectChildName
-import akka.actor.SelectParent
 import akka.actor.SelectChildPattern
+import akka.actor.SelectParent
+import akka.remote.DaemonMsgCreate
+import akka.serialization.SerializationExtension
+import akka.testkit.AkkaSpec
+import akka.testkit.TestActors
 
 class MessageContainerSerializerSpec extends AkkaSpec {
 
@@ -35,8 +37,17 @@ class MessageContainerSerializerSpec extends AkkaSpec {
           wildcardFanOut = true))
     }
 
-    def verifySerialization(msg: AnyRef): Unit = {
-      ser.deserialize(ser.serialize(msg).get, msg.getClass).get should ===(msg)
+    "serialize and deserialize DaemonMsgCreate with tagged actor" in {
+      val props = TestActors.echoActorProps.withActorTags("one", "two")
+      val deserialized =
+        verifySerialization(DaemonMsgCreate(props, props.deploy, "/user/some/path", system.deadLetters))
+      deserialized.deploy.tags should ===(Set("one", "two"))
+    }
+
+    def verifySerialization[T <: AnyRef](msg: T): T = {
+      val deserialized = ser.deserialize(ser.serialize(msg).get, msg.getClass).get
+      deserialized should ===(msg)
+      deserialized
     }
 
   }