zio / zio-akka-cluster Goto Github PK

ZIO wrapper for Akka Cluster

Home Page: https://zio.dev/zio-akka-cluster

License: Apache License 2.0

Scala 100.00%

zio-akka-cluster's Introduction

Project Stage	CI	Release	Snapshot	Issues

Scaladoc	Scaladex	Discord	Twitter	Gitpod
Scaladoc

Welcome to ZIO

ZIO is a zero-dependency Scala library for asynchronous and concurrent programming.

Powered by highly-scalable, non-blocking fibers that never waste or leak resources, ZIO lets you build scalable, resilient, and reactive applications that meet the needs of your business.

High-performance. Build scalable applications with minimal runtime overhead.
Type-safe. Use the full power of the Scala compiler to catch bugs at compile time.
Concurrent. Easily build concurrent apps without deadlocks, race conditions, or complexity.
Asynchronous. Write sequential code that looks the same whether it's asynchronous or synchronous.
Resource-safe. Build apps that never leak resources (including threads!), even when they fail.
Testable. Inject test services into your app for fast, deterministic, and type-safe testing.
Resilient. Build apps that never lose errors, and which respond to failure locally and flexibly.
Functional. Rapidly compose solutions to complex problems from simple building blocks.

To learn more about ZIO, see the following references:

Adopters

Following is a partial list of companies happily using ZIO in production to craft concurrent applications.

Want to see your company here? Submit a PR!

AdGear / Samsung Ads
Adidas
adpulse.io
adsquare
Amazon
Anduin Transactions
AnyMind Group
Ayolab
Asana
Aurinko
auto.ru
AutoScout24
Avast
Bank of America
Believe
Blue Insight Digital
BlueLabs
Bpp
Broad
Caesars Digital
CalcBank
Call Handling
Carvana
Cellular
ChannelPilot
Chartboost
Cloudfarms
CodeComprehension
Codept
Colisweb
Collibra
Compellon
Complicated Robot
Conduktor
Contramap
Coralogix
Credit Karma
CurrencyCloud
D.E.Solution
DaSCH
DataChef
Demandbase
Demyst
Devsisters
DHL Parcel The Netherlands
Disney+ Streaming
Doomoolmori
Dow Jones
DPG recruitment
Dream11
Deutsche Telekom IoT GmbH
eBay
Eaglescience
Electricité de France (EDF)
EnelX
Evolution
Evo
Flipp
Fugo.ai
Garner Distributed Workflow
Glean
GrandParade
greyflower.media GmbH
Hunters.AI
Hypefactors
iHeartRadio
IHS Markit
Investsuite
Invia
Kaizen Solutions
Kamon APM
Kodmagi
Kensu
LambdaWorks
LeadIQ
Lernkunst
LiveIntent Inc.
Lottoland
MATECHS
Megogo
MOIA
Mylivn
NewMotion
Nexxchange
Nike
NsLookup
Ocado Technology
Olyro GmbH
Optrak
Performance Immo
Pirum
PITS Global Data Recovery Services
Playtika
PPC Samurai
Prezi
Radix Labs
Railroad19
Randstad Groep Nederland
Rapidor
PIM Solutions
Remember The Milk
REWE Digital
LKWfee
Risk Ident
Rocker
Rudder
Sanjagh
Scalac
SecurityScorecard
SentinelOne
Signicat
Société Générale Corporate and Investment Banking
SoftwareMill
StreamWeaver
Stuart
Teads
The Pokemon Company International
TomTom
Tinka
Tinkoff
Trackabus
Trainor
Tranzzo
TreuTech
Tweddle Group
Undo
Unit
Univalence
Unzer
Vakantiediscounter
Verbund AG
Waylay
Wehkamp
Wefunder
Wolt
Yandex.Classifieds
Audela
Valamis Group
Valsea
VirtusLab
Vish
Vivid Money
Zalando
Zooz

Learn More on the ZIO Homepage

Code of Conduct

See the Code of Conduct

Support

Come chat with us on .

Legal

zio-akka-cluster's People

Contributors

Stargazers

Watchers

zio-akka-cluster's Issues

Race condition between creation of ActorSystem and Cluster.clusterEvents

Hi again,

I think I´m facing a possible Race condition using


  val clusterEvents: ZIO[ActorSystem, Throwable, Queue[ClusterDomainEvent]] =
    Queue.unbounded[ClusterDomainEvent].tap(clusterEventsWith)

This queue is filled with the events of actions over the cluster.

The problem is that the ActorSystem has been created previously passed as provider into the Monad ZIO[ActorSystem, Throwable, Any]

mainProgram.unsafeRunToFuture(bootableProgram.provide(actorSystem))

so it happening sometimes that the Member is added into the cluster before I subscribe into the Queue so I´m missing that event.

events <- Cluster.clusterEvents
event <-events.take

Any suggestion to prevent that RC?

I guess not pass the ActorSystem as Environment when I run the monad and pass into the Cluster.clusterEvents, is an option but definitely less cool option :(

 events <- Cluster.clusterEvents.provide(ActorSystem.create("bla"))
event <-events.take

Regards.

No communication between nodes.

I´m trying to send a message from one ActorSystem to another.

The publisher in port 2700 and listener in 2600.

But checking the logs it seems it´s trying to publish the message from the listener port 2600 to another actor without port Actor[akka://ClusterSystem/system/distributedPubSubMediator]

[INFO] [07/18/2019 16:26:55.893] [ClusterSystem-akka.actor.default-dispatcher-3] [akka://ClusterSystem/system/distributedPubSubMediator] Message [akka.cluster.pubsub.DistributedPubSubMediator$Internal$Status] from Actor[akka.tcp://[email protected]:2600/system/distributedPubSubMediator#-327490257] to Actor[akka://ClusterSystem/system/distributedPubSubMediator] was not delivered. [2] dead letters encountered. If this is not an expected behavior, then [Actor[akka://ClusterSystem/system/distributedPubSubMediator]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

[INFO] [07/18/2019 16:26:55.893] [ClusterSystem-akka.actor.default-dispatcher-3] [akka://ClusterSystem/system/distributedPubSubMediator] Message [akka.cluster.pubsub.DistributedPubSubMediator$Internal$Status] from Actor[akka.tcp://[email protected]:2600/system/distributedPubSubMediator#-327490257] to Actor[akka://ClusterSystem/system/distributedPubSubMediator] was not delivered. [2] dead letters encountered. If this is not an expected behavior, then [Actor[akka://ClusterSystem/system/distributedPubSubMediator]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [07/18/2019 16:26:56.883] [ClusterSystem-akka.actor.default-dispatcher-23] [akka://ClusterSystem/system/distributedPubSubMediator] Message [akka.cluster.pubsub.DistributedPubSubMediator$Internal$Status] from Actor[akka.tcp://[email protected]:2600/system/distributedPubSubMediator#-327490257] to Actor[akka://ClusterSystem/system/distributedPubSubMediator] was not delivered. [3] dead letters encountered. If this is not an expected behavior, then [Actor[akka://ClusterSystem/system/distributedPubSubMediator]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [07/18/2019 16:26:57.872] [ClusterSystem-akka.actor.default-dispatcher-23] [akka://ClusterSystem/system/distributedPubSubMediator] Message [akka.cluster.pubsub.DistributedPubSubMediator$Internal$Status] from Actor[akka.tcp://[email protected]:2600/system/distributedPubSubMediator#-327490257] to Actor[akka://ClusterSystem/system/distributedPubSubMediator] was not delivered. [4] dead letters encountered. If this is not an expected behavior, then [Actor[akka://ClusterSystem/system/distributedPubSubMediator]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [07/18/2019 16:26:58.896] [ClusterSystem-akka.actor.default-dispatcher-3] [akka://ClusterSystem/system/distributedPubSubMediator] Message [akka.cluster.pubsub.DistributedPubSubMediator$Internal$Status] from Actor[akka.tcp://[email protected]:2600/system/distributedPubSubMediator#-327490257] to Actor[akka://ClusterSystem/system/distributedPubSubMediator] was not delivered. [5] dead letters encountered. If this is not an expected behavior, then [Actor[akka://ClusterSystem/system/distributedPubSubMediator]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [07/18/2019 16:26:59.877] [ClusterSystem-akka.actor.default-dispatcher-24] [akka://ClusterSystem/system/distributedPubSubMediator] Message [akka.cluster.pubsub.DistributedPubSubMediator$Internal$Status] from Actor[akka.tcp://[email protected]:2600/system/distributedPubSubMediator#-327490257] to Actor[akka://ClusterSystem/system/distributedPubSubMediator] was not delivered. [6] dead letters encountered. If this is not an expected behavior, then [Actor[akka://ClusterSystem/system/distributedPubSubMediator]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[WARN] [SECURITY][07/18/2019 16:27:00.042] [pool-2-thread-2] [akka.serialization.Serialization(akka://ClusterSystem)] Using the default Java serializer for class [zio.akka.cluster.pubsub.MessageEnvelope] which is not recommended because of performance implications. Use another serializer or disable this warning using the setting 'akka.actor.warn-about-java-serializer-usage'
[INFO] [07/18/2019 16:27:00.047] [ClusterSystem-akka.actor.default-dispatcher-3] [akka://ClusterSystem/system/distributedPubSubMediator] Message [zio.akka.cluster.pubsub.MessageEnvelope] without sender to Actor[akka://ClusterSystem/system/distributedPubSubMediator#2077152297] was not delivered. [7] dead letters encountered. If this is not an expected behavior, then [Actor[akka://ClusterSystem/system/distributedPubSubMediator#2077152297]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

Any idea if I´m doing something wrong?, here the code

 val messageProgram: ZIO[ActorSystem, Throwable, String] = (for {
        remotePort <- ZIO.accessM[ActorSystem](as => ZIO.effect(as.settings.config.getString("akka.remote.netty.tcp.port")))
        state <- Cluster.clusterState
        _ <- logClusterInfo(remotePort, state)
        _ <- if (state.members.size > 1) {
          //Wait until the other node accept the new member
          Thread.sleep(5000)
          for {
            pubSub <- PubSub.createPubSub[String]
            _ <- pubSub.publish(ToggleTopic, "Hello ZIO world")
          } yield Unit
        }else{
          ZIO.effect(Unit)
        }
        subscriber <- PubSub.createSubscriber[String]
        queue <- subscriber.listen(ToggleTopic)
        message <- queue.take
      } yield message).catchAll(t => {
        logger.info(null, "Error in ZIO Akka cluster:")
        ZIO.fail(t)
      })

As far as I can see in the akka log, the cluster is formed

Members:TreeSet(Member(address = akka.tcp://[email protected]:2600, status = Up), Member(address = akka.tcp://[email protected]:2700, status = Up))

Any idea what it could be wrong?
Regards

ZIO is at Milestone 4, with an RC expected in the next few weeks.
https://github.com/zio/zio/releases/tag/v2.The API is nearly stable at this point, so any early migration work against this version should pay off towards the official 2.0 release.

The progress is being tracked here:
zio/zio#5470

The Stream Encoding work in progress is the only area where the API might still change before the RC.

To assist with the migration, follow this guide:
https://zio.dev/howto/migrate/zio-2.x-migration-guide/

We are actively working on a ScalaFix rule that will cover the bulk of the simple API changes:
https://github.com/zio/zio/blob/series/2.x/scalafix/rules/src/main/scala/fix/Zio2Upgrade.scala

If you would like assistance with the migration from myself or other ZIO contributors, please let us know!

onMessage: Msg => ZIO[Entity[State], Nothing, Unit]

in Sharding.start to accept more environment, and provide it from outside like

Sharding.start(...).provideLayer(layerForOnMessage)`

How to run the ZIO

I´m trying to use the library, but I´m not able to run the ZIO.

Here my code:

   val main: DefaultRuntime = new DefaultRuntime {}

      val clusterStateProgram: ZIO[ActorSystem, Throwable, CurrentClusterState] =
        for {
          state <- Cluster.clusterState
          state <- logClusterInfo(state)
        } yield state

      main.unsafeRun(clusterStateProgram)

and the compilation error

Error:(64, 22) type mismatch;
 found   : zio.ZIO[akka.actor.ActorSystem,Throwable,akka.cluster.ClusterEvent.CurrentClusterState]
 required: zio.ZIO[zio.clock.Clock with zio.console.Console with zio.system.System with zio.random.Random with zio.blocking.Blocking,?,?]
      main.unsafeRun(clusterStateProgram)

Add 2.13 build/CI

ZIO now support Scala 2.13, so this library should too.