Add support for querying last known sequence number by persistenceId. #267
Conversation
core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/DefaultReadJournalDao.scala
Outdated
Show resolved
Hide resolved
core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/ReadJournalQueries.scala
Outdated
Show resolved
Hide resolved
core/src/main/scala/org/apache/pekko/persistence/jdbc/query/javadsl/JdbcReadJournal.scala
Outdated
Show resolved
Hide resolved
|
@pjfanning I tried to align column widths to 120 and dropped curly braces for single statement methods. I hope that fits better now. |
| @@ -0,0 +1,27 @@ | |||
| /* | |||
There was a problem hiding this comment.
Can you change the license header on new files that are not copies of other files to be https://github.com/apache/pekko/blob/main/project/AddMetaInfLicenseFiles.scala#L1-L16 ?
There was a problem hiding this comment.
Sure! Updated in b033218.
| import org.scalatest.concurrent.ScalaFutures | ||
|
|
||
| abstract class CurrentLastKnownSequenceNumberByPersistenceIdTest( | ||
| config: String |
There was a problem hiding this comment.
this style is still wrong
There was a problem hiding this comment.
Hope f0cc6c5 does the job?
| @@ -32,7 +38,8 @@ class JdbcReadJournal(journal: ScalaJdbcReadJournal) | |||
| with CurrentEventsByPersistenceIdQuery | |||
| with EventsByPersistenceIdQuery | |||
| with CurrentEventsByTagQuery | |||
| with EventsByTagQuery { | |||
| with EventsByTagQuery | |||
| with CurrentLastKnownSequenceNumberByPersistenceIdQuery { | |||
There was a problem hiding this comment.
Do we really need the CurrentLastKnownSequenceNumberByPersistenceIdQuery trait? Can't we just add the function without having a marker trait.
We can add the trait to the pekko-persistence jar at some stage and then do another PR to uptake the trait here.
There was a problem hiding this comment.
Assuming a marker trait would allow writing this:
PersistenceQuery(system).readJournalFor[CurrentLastKnownSequenceNumberByPersistenceIdQuery](JdbcReadJournal.Identifier)
Instead of having to write this:
PersistenceQuery(system).readJournalFor[JdbcReadJournal](JdbcReadJournal.Identifier)
But if the goal is to eventually have the marker trait in the core repository then this serves very little and temporary purpose, so overall I agree with suggestion to remove it.
There was a problem hiding this comment.
Main intent was to stick to existing marker traits application + having that part of contract testable on its own. In our applications we rarely declare dependencies to the full query journal, usually we tend to have the capabilities speaking:
object FancyAggregator {
def apply(readJournal: EventsByTagQuery)
}vs
object FancyAggregator {
def apply(readJournal: JdbcReadJournal)
}If you feel more comfortable, surely, I can try to inline it without marker trait. Would make dependency injection a bit harder.
There was a problem hiding this comment.
My aim would be merge this without the marker trait meaning that we could release a version pekko-persistence-jdbc 1.1.x that supports this but without needing a new Pekko core release. pekko-persistence 1.2.0-M2 and above could have the marker trait and a pekko-persistence-jdbc 1.2.0(-M1) release could rely on pekko-persistence 1.2.0-M2 and uptake the marker trait.
I haven't studied pekko-persistence-r2dbc but I presume that it too could implement this support and in a similar timeline.
There was a problem hiding this comment.
Sure, I will drop the marker trait today or tomorrow.
There was a problem hiding this comment.
Removed marker trait with ce15d5e.
There was a problem hiding this comment.
@ptrdom it looks like pekko-persistence-r2dbc can have a similar PR added at some stage? Can you confirm this? I can look into adding the marker trait to the pekko core but it could be a while before it makes it into a full release meaning that we will have to wait to add the uptake of the marker trait.
There was a problem hiding this comment.
Yes, I think it should be no problem to add this functionality to pekko-persistence-r2dbc.
| * @param persistenceId The `persistenceId` for which the last known sequence number should be returned. | ||
| * @return Some sequence number or None if the `persistenceId` is unknown. | ||
| */ | ||
| def currentLastKnownSequenceNumberByPersistenceId(persistenceId: String): CompletionStage[Optional[Long]] = |
There was a problem hiding this comment.
There was a problem hiding this comment.
I was considering scala.Long as the better suited one here, since the EventEnvelope itself - which is the return type of multiple other methods in this class - carries the offset and specially the sequenceNr - the attribute that this method is actually interested in - as scala.Long.
The Optional wrapper was "inspired" by the GetObjectResult for the DurableStateStore javadsl.
There was a problem hiding this comment.
Quick search in Pekko repository gives mixed results, some javadsl classes stick to java.lang.Long quite well, others do not. My guess is that javadsl should use java.lang.Long, it is just difficult to enforce, because there is no automatic linting for it.
There was a problem hiding this comment.
I agree that Pekko is inconsistent but I would prefer that we use java.lang.Long here.
Resolves apache/pekko#1761 for this plugin.