feat: Add LatestEventTimestampQuery (#32700)

* Retrieve the latest timestamp for an entity type and slice range.
* Primary purpose is for observability.

Author: patriknw

akka-persistence-query/src/main/scala/akka/persistence/query/typed/javadsl/EventsBySliceFirehoseQuery.scala

@@ -48,7 +48,8 @@ final class EventsBySliceFirehoseQuery(delegate: scaladsl.EventsBySliceFirehoseQ
     with EventsBySliceQuery
     with EventsBySliceStartingFromSnapshotsQuery
     with EventTimestampQuery
-    with LoadEventQuery {
+    with LoadEventQuery
+    with LatestEventTimestampQuery {
 
   override def sliceForPersistenceId(persistenceId: String): Int =
     delegate.sliceForPersistenceId(persistenceId)
@@ -82,4 +83,10 @@ final class EventsBySliceFirehoseQuery(delegate: scaladsl.EventsBySliceFirehoseQ
   override def loadEnvelope[Event](persistenceId: String, sequenceNr: Long): CompletionStage[EventEnvelope[Event]] =
     delegate.loadEnvelope[Event](persistenceId, sequenceNr).asJava
 
+  override def latestEventTimestamp(
+      entityType: String,
+      minSlice: Int,
+      maxSlice: Int): CompletionStage[Optional[Instant]] =
+    delegate.latestEventTimestamp(entityType, minSlice, maxSlice).map(_.toJava)(ExecutionContext.parasitic).asJava
+
 }

akka-persistence-query/src/main/scala/akka/persistence/query/typed/javadsl/LatestEventTimestampQuery.scala

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2021-2025 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.persistence.query.typed.javadsl
+
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.CompletionStage
+
+import akka.persistence.query.javadsl.ReadJournal
+
+/**
+ * Retrieve the latest timestamp for an entity type and slice range.
+ */
+trait LatestEventTimestampQuery extends ReadJournal {
+
+  def latestEventTimestamp(entityType: String, minSlice: Int, maxSlice: Int): CompletionStage[Optional[Instant]]
+
+}

akka-persistence-query/src/main/scala/akka/persistence/query/typed/scaladsl/EventsBySliceFirehoseQuery.scala

@@ -50,7 +50,8 @@ final class EventsBySliceFirehoseQuery(system: ExtendedActorSystem, config: Conf
     with EventsBySliceQuery
     with EventsBySliceStartingFromSnapshotsQuery
     with EventTimestampQuery
-    with LoadEventQuery {
+    with LoadEventQuery
+    with LatestEventTimestampQuery {
 
   private lazy val persistenceExt = Persistence(system)
   private lazy val settings = EventsBySliceFirehose.Settings(system, cfgPath)
@@ -100,6 +101,15 @@ final class EventsBySliceFirehoseQuery(system: ExtendedActorSystem, config: Conf
           "doesn't implement LoadEventQuery")
     }
 
+  override def latestEventTimestamp(entityType: String, minSlice: Int, maxSlice: Int): Future[Option[Instant]] = {
+    eventsBySliceQuery match {
+      case q: LatestEventTimestampQuery => q.latestEventTimestamp(entityType, minSlice, maxSlice)
+      case _ =>
+        throw new IllegalArgumentException(
+          s"Underlying ReadJournal [${settings.delegateQueryPluginId}] doesn't implement LatestEventTimestampQuery")
+    }
+  }
+
   private def eventsBySliceQuery: EventsBySliceQuery = {
     val delegateQueryPluginId =
       EventsBySliceFirehose.Settings.delegateQueryPluginId(system.settings.config.getConfig(cfgPath))

akka-persistence-query/src/main/scala/akka/persistence/query/typed/scaladsl/LatestEventTimestampQuery.scala

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2021-2025 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.persistence.query.typed.scaladsl
+
+import java.time.Instant
+
+import scala.concurrent.Future
+
+import akka.persistence.query.scaladsl.ReadJournal
+
+/**
+ * Retrieve the latest timestamp for an entity type and slice range.
+ */
+trait LatestEventTimestampQuery extends ReadJournal {
+
+  def latestEventTimestamp(entityType: String, minSlice: Int, maxSlice: Int): Future[Option[Instant]]
+
+}