Offset batch: remove intermediary structure (#870)

akka · Aug 23, 2019 · 05ecc0c · 05ecc0c
1 parent c2b9470
commit 05ecc0c
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 26 deletions.
diff --git a/build.sbt b/build.sbt
@@ -100,7 +100,7 @@ val commonSettings = Def.settings(
         Seq(
           "-doc-source-url", {
             val branch = if (isSnapshot.value) "master" else s"v${version.value}"
-            s"https://github.com/akka/alpakka-kafka/tree/${branch}€{FILE_PATH_EXT}#€{FILE_LINE}"
+            s"https://github.com/akka/alpakka-kafka/tree/${branch}€{FILE_PATH_EXT}#L€{FILE_LINE}"
           },
           "-doc-canonical-base-url",
           "https://doc.akka.io/api/alpakka-kafka/current/"

diff --git a/core/src/main/scala/akka/kafka/ConsumerMessage.scala b/core/src/main/scala/akka/kafka/ConsumerMessage.scala
@@ -12,6 +12,7 @@ import akka.Done
 import akka.annotation.{DoNotInherit, InternalApi}
 import akka.kafka.internal.{CommittableOffsetBatchImpl, CommittedMarker}
 import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.kafka.common.TopicPartition
 
 import scala.concurrent.Future
 import scala.runtime.AbstractFunction2
@@ -137,7 +138,9 @@ object ConsumerMessage {
       groupId: String,
       topic: String,
       partition: Int
-  )
+  ) {
+    def topicPartition: TopicPartition = new TopicPartition(topic, partition)
+  }
 
   object CommittableOffsetBatch {
     val empty: CommittableOffsetBatch = new CommittableOffsetBatchImpl(Map.empty, Map.empty, 0)

diff --git a/core/src/main/scala/akka/kafka/internal/CommittableSources.scala b/core/src/main/scala/akka/kafka/internal/CommittableSources.scala
@@ -7,12 +7,7 @@ package akka.kafka.internal
 
 import akka.actor.ActorRef
 import akka.annotation.InternalApi
-import akka.kafka.ConsumerMessage.{
-  CommittableMessage,
-  CommittableOffset,
-  CommittableOffsetBatch,
-  PartitionOffsetMetadata
-}
+import akka.kafka.ConsumerMessage.{CommittableMessage, CommittableOffset, CommittableOffsetBatch, GroupTopicPartition}
 import akka.kafka._
 import akka.kafka.internal.KafkaConsumerActor.Internal.{Commit, CommitSingle}
 import akka.kafka.scaladsl.Consumer.Control
@@ -26,7 +21,6 @@ import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, Offset
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.requests.OffsetFetchResponse
 
-import scala.collection.immutable
 import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{ExecutionContext, Future}
 
@@ -139,26 +133,20 @@ private[kafka] class KafkaAsyncConsumerCommitterRef(consumerActor: ActorRef, com
       )
     )
 
-  private def commit(offsets: immutable.Seq[PartitionOffsetMetadata]): Future[Done] = {
-    val offsetsMap: Map[TopicPartition, OffsetAndMetadata] = offsets.map { offset =>
-      new TopicPartition(offset.key.topic, offset.key.partition) ->
-      new OffsetAndMetadata(offset.offset + 1, offset.metadata)
-    }.toMap
-    sendCommit(Commit(offsetsMap))
-  }
-
   def commit(batch: CommittableOffsetBatch): Future[Done] = batch match {
     case b: CommittableOffsetBatchImpl =>
-      val futures = b.offsetsAndMetadata.groupBy(_._1.groupId).map {
+      val groupIdOffsetMaps: Map[String, Map[GroupTopicPartition, OffsetAndMetadata]] =
+        b.offsetsAndMetadata.groupBy(_._1.groupId)
+      val futures = groupIdOffsetMaps.map {
         case (groupId, offsetsMap) =>
           val committer = b.committers.getOrElse(
             groupId,
             throw new IllegalStateException(s"Unknown committer, got [$groupId]")
           )
-          val offsets: immutable.Seq[PartitionOffsetMetadata] = offsetsMap.map {
-            case (ctp, offset) => PartitionOffsetMetadata(ctp, offset.offset(), offset.metadata())
-          }.toList
-          committer.commit(offsets)
+          val offsets: Map[TopicPartition, OffsetAndMetadata] = offsetsMap.map {
+            case (gtp, offset) => gtp.topicPartition -> offset
+          }
+          committer.sendCommit(Commit(offsets))
       }
       Future.sequence(futures).map(_ => Done)
 
@@ -169,7 +157,7 @@ private[kafka] class KafkaAsyncConsumerCommitterRef(consumerActor: ActorRef, com
       )
   }
 
-  private[this] def sendCommit(msg: AnyRef): Future[Done] = {
+  private def sendCommit(msg: AnyRef): Future[Done] = {
     import akka.pattern.ask
     consumerActor
       .ask(msg)(Timeout(commitTimeout))

diff --git a/core/src/main/scala/akka/kafka/internal/MessageBuilder.scala b/core/src/main/scala/akka/kafka/internal/MessageBuilder.scala
@@ -162,7 +162,7 @@ private[kafka] final class CommittableOffsetBatchImpl(
     val committers: Map[String, KafkaAsyncConsumerCommitterRef],
     override val batchSize: Long
 ) extends CommittableOffsetBatch {
-  def offsets: Map[GroupTopicPartition, Long] = offsetsAndMetadata.view.mapValues(_.offset()).toMap
+  def offsets: Map[GroupTopicPartition, Long] = offsetsAndMetadata.view.mapValues(_.offset() - 1L).toMap
 
   def updated(committable: Committable): CommittableOffsetBatch = committable match {
     case offset: CommittableOffset => updatedWithOffset(offset)
@@ -180,7 +180,7 @@ private[kafka] final class CommittableOffsetBatchImpl(
     }
 
     val newOffsets =
-      offsetsAndMetadata.updated(key, new OffsetAndMetadata(committableOffset.partitionOffset.offset, metadata))
+      offsetsAndMetadata.updated(key, new OffsetAndMetadata(committableOffset.partitionOffset.offset + 1L, metadata))
 
     val committer = committableOffset match {
       case c: CommittableOffsetImpl => c.committer
@@ -238,7 +238,7 @@ private[kafka] final class CommittableOffsetBatchImpl(
     offsets.asJava
 
   override def toString(): String =
-    s"CommittableOffsetBatch(batchSize=$batchSize, ${offsetsAndMetadata.mkString("->")})"
+    s"CommittableOffsetBatch(batchSize=$batchSize, ${offsets.mkString(", ")})"
 
   override def commitScaladsl(): Future[Done] =
     if (batchSize == 0L)