safety-data / akka-persistence-redis Goto Github PK

When I configure the plugin to use sentinel, I get failures on journal queries. The problem seems to be that in EventsByPersistenceId, EventsByTagSource, and PersistenceIdsSource, the code in preStart() to create the subscription assumes the standard config, rather than sentinel. I have it working by changing:

        subscription = RedisPubSub(host = RedisUtils.host(conf),
          port = RedisUtils.port(conf),
          channels = Seq(identifiersChannel),
          patterns = Nil,
          authPassword = RedisUtils.password(conf),
          onMessage = callback.invoke)(system)

to

        subscription = RedisPubSub(host = redis.host,
          port = redis.port,
          channels = Seq(identifiersChannel),
          patterns = Nil,
          authPassword = redis.password,
          onMessage = callback.invoke)(system)

Does that fix seem reasonable?

Thanks -

Is this private[redis] intentional? I'm attempting to stub out my read journal in my application tests but this seems to be preventing me from doing so.

[error] AppActorTest.scala:16: constructor ScalaReadJournal in class ScalaReadJournal cannot be accessed in <$anon: akka.persistence.query.journal.redis.ScalaReadJournal>
[error]   val stubbedReadJournal: ScalaReadJournal = new ScalaReadJournal {
[error]                                                  ^
[error] one error found

Database configuration should allow for configuring Redis Sentinel.

redis.actors.ReplyErrorException: NOAUTH Authentication required.

[ERROR] [10/17/2019 18:08:38.511] [alpha-akka-akka.actor.default-dispatcher-4] [akka://alpha-akka/user/b60bc42a-f8e3-4227-a911-a82f65a78d21] Persistence failure when replaying events for persistenceId [b60bc42a-f8e3-4227-a911-a82f65a78d21]. Last known sequence number [0]
redis.actors.ReplyErrorException: NOAUTH Authentication required.
at redis.actors.RedisReplyDecoder.$anonfun$decodeRedisReply$1(RedisReplyDecoder.scala:68)
at redis.actors.RedisReplyDecoder.$anonfun$decodeRedisReply$1$adapted(RedisReplyDecoder.scala:67)
at redis.protocol.DecodeResult.foreach(RedisProtocolReply.scala:89)
at redis.protocol.DecodeResult.foreach$(RedisProtocolReply.scala:87)
at redis.protocol.FullyDecoded.foreach(RedisProtocolReply.scala:112)
at redis.actors.RedisReplyDecoder.decodeRedisReply(RedisReplyDecoder.scala:67)
at redis.actors.RedisReplyDecoder.decodeRepliesRecur(RedisReplyDecoder.scala:50)
at redis.actors.RedisReplyDecoder.decodeReplies(RedisReplyDecoder.scala:35)
at redis.actors.RedisReplyDecoder$$anonfun$receive$1.applyOrElse(RedisReplyDecoder.scala:28)
at akka.actor.Actor.aroundReceive(Actor.scala:517)
at akka.actor.Actor.aroundReceive$(Actor.scala:515)
at redis.actors.RedisReplyDecoder.aroundReceive(RedisReplyDecoder.scala:11)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:588)
at akka.actor.ActorCell.invoke(ActorCell.scala:557)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
at akka.dispatch.Mailbox.run(Mailbox.scala:225)
at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

redis config
akkaConfig: akka.persistence.journal.plugin: akka-persistence-redis.journal akka.persistence.snapshot-store.plugin: akka-persistence-redis.snapshot akka-persistence-redis: redis: mode: sentinel database: 1 password: 123456 master: mymaster sentinel-list: 127.0.0.1:26379 #host1:1234,host2:1235

This is very strange :

  private val config = ConfigFactory.parseString(
    """
        akka {
          persistence.journal.plugin = "akka-persistence-redis.journal"
        }
        akka-persistence-redis {
          redis {
            host = redis
          }
        }
        """.stripMargin
  ).withFallback(ConfigFactory.load())

  val system = ActorSystem("example", config)

 Connect to localhost/127.0.0.1:6379
[akka://example/user/RedisClient-$a] CommandFailed(Connect(localhost/127.0.0.1:6379,None,List(KeepAlive(true)),None,false)) because of Connection refused

I tried to expose even redis.host = redis to the global scope, but I just cannot make it to connect to hostname redis. Any idea?

Added support through this pr: #27

Scala 2.13.0 release notes: https://github.com/scala/scala/releases/tag/v2.13.0

Hey,

would you please add a paragraph about why should people use this over hootsuite/akka-persistence-redis ?

I'm currently using the above mentioned plugin but for some reason ~ 1 of 1M messages fails to be persisted and I cannot figure out why, so I'm wondering whether to switch to this plugin implementation.

I've seen this error appear transiently in my logs when using eventsByPersistenceId:

Unexpected source state when querying: 2

I think it tracks down to

which seems like it can get hit by

Am I following this code path correctly?

Would it be possible to release a new version with 2.5.17 akka ?

Publish artifacts on maven central repository and document it.

May be, this is more of a question rather than issue. Is there any plans to support competing consumers part of persistenceQuery plugin?

Below is a quick peek of what I'm trying to do...When I run multiple instances of the "persistence reader actor", all the actors (# 6 below) receive the same message. Ideally, I would want one of the actors to receive the message vs all. Is it possible to implement "competing consumers" so that I can avoid the same message being processed by multiple persistence reader actors?

I did go thru akka "smallest-mailbox-pool" to implement competing consumers but wanted to make sure that there is nothing inbuilt in persistenceQuery plugin before handling competing consumers with my own code.

Or please suggest if there is a better design for this..

Hi,

Below is my example config, I'm connecting to remote redis server. It successfully connects to remote server, regardless, its always looking for localhost:6379 too and erroring out "connection refused"..

how can I solve this ?

Please point me to documentation, if any.

    akka-persistence-redis {
      redis {
        mode = simple
        host = blah.redislabs.com
        port = 13423
        password = blah
     }
     journal.redis = ${akka-persistence-redis.redis}
   }

a green cucumber should be sent before line 111, not after
https://github.com/safety-data/akka-persistence-redis/blob/master/src/test/scala/akka/persistence/query/journal/redis/EventsByTagSpec.scala#L117

Right now currentEventsByTag query could fetch the data from Redis more than one time

Current implementation call some stage method, in particular completeStage and failStage in a Future.onComplete instead of a dedicated stage callback.

This results in unexpected results and unreliable implementation and does not respect the custom stage implementation guidelines.

Make sure that all stage methods are called in a proper callback.

Please, contribute akka 2.6.X

I'm trying to call eventsByPersistenceId(id, 0L, Long.MaxValue) with a single event in my journal that has a score of 600. The reason for this large range is to effectively say I want every event for a given persistenceId.

It appears that the following code makes the zrangebyscore call with a with a min score of 0 and max score of 100 (the default configured max value). Is this intentional?

Publish scaladoc on github pages and link to them in the README.

instead of
akka.persistence.snapshot.plugin = "akka-persistence-redis.snapshot"
it rather seems to be
akka.persistence.snapshot-store.plugin = "akka-persistence-redis.snapshot"

Each asyncWriteOperation stores a tag into tagsKey, but these values are not used anywhere