Akka cluster node unreachable when update distribute data in different nodes

1

I create a cluster with 5nodes，when I update ORSetKey(about 1000 times at the same time)，the nodes will lost，this is the warnning info:
2022-06-06 18:00:44,941 WARN akka.cluster.ClusterHeartbeat - Cluster Node [akka.tcp://datasync-service-application@127.0.0.1:2551] - Scheduled sending of heartbeat was delayed. Previous heartbeat was sent [2194] ms ago, expected interval is [1000] ms. This may cause failure detection to mark members as unreachable. The reason can be thread starvation, e.g. by running blocking tasks on the default dispatcher, CPU overload, or GC.

akka version 2.6.17

application.conf

akka.cluster.jmx.multi-mbeans-in-same-jvm = on
akka.remote.artery.enabled=false

akka {
  cluster.roles = ["worker-node"]
  actor {
    provider = "cluster"
  }
  remote.classic {
    log-remote-lifecycle-events = off
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2551
      maximum-frame-size=328000b
    }
  }

  cluster {
    seed-nodes = ["akka.tcp://datasync-service-application@127.0.0.1:2551"]
  }
  discovery {
    method = akka-dns
  }
}

main class

object BalanceTest extends App{
  val logger = Logger(this.getClass)

  def create(port:Int) ={
    val config = ConfigFactory.parseString(s"akka.remote.classic.netty.tcp.port=$port")
      .withFallback(ConfigFactory.load("test"))
    implicit val system = ActorSystem("datasync-service-application",config)


    val workerRouter: ActorRef = {
      val routerProps = ClusterRouterGroup(
        RoundRobinGroup(List("/user/WorkerActor")),
        ClusterRouterGroupSettings(
          totalInstances = 1000,
          routeesPaths = List("/user/WorkerActor"),
          allowLocalRoutees = true,
          useRoles = Set("worker-node")
        )
      ).props
      system.actorOf(routerProps, "workerRouter")
    }

    system.actorOf(Props(classOf[WorkerActor]), "WorkerActor")

    workerRouter
  }




  val ref1: ActorRef = create(2551)
  val ref2: ActorRef = create(2552)
  val ref3: ActorRef = create(2553)
  val ref4: ActorRef = create(2554)
  val ref5: ActorRef = create(2555)

  Thread.sleep(3000)
  for(i<-1 to 30000){
    ref1! s"h$i"
  }

}

actor

class WorkerActor extends Actor{

  lazy val address = Cluster(context.system).selfAddress.toString
  val replicator :ActorRef =  DistributedData(context.system).replicator
  implicit val cluster = Cluster(context.system)
  protected implicit lazy val ec: ExecutionContext = context.dispatcher

  import EventTest._
  override def preStart(): Unit ={
    replicator ! Subscribe(Flags,self)
    println("started")
  }


  override def receive: Receive = {
    case jobId:String =>
      replicator ! Update(EventTest.Flags, ORSet(), EventTest.writeAll)(_+jobId)
    case c @ Changed(Flags) =>
      val dData = c.get(Flags).elements
      println(dData.size)
  }
}