kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject kafka git commit: KAFKA-5247; Materialize committed offsets in offset order
Date Tue, 23 May 2017 16:47:07 GMT
Repository: kafka
Updated Branches:
  refs/heads/0.11.0 33216846f -> 2b326faf5


KAFKA-5247; Materialize committed offsets in offset order

With this patch, offset commits are always materialized according to the order of the commit
records in the offsets topic.

Before this patch, transactional offset commits were materialized in transaction order. However,
the log cleaner will always preserve the record with the greatest offset. This meant that
if there was a mix of offset commits from a consumer and a transactional producer, then it
we would switch from transactional order to offset order after cleaning, resulting in an inconsistent
state.

Author: Apurva Mehta <apurva@confluent.io>

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Jason Gustafson <jason@confluent.io>

Closes #3108 from apurvam/KAFKA-5247-materialize-committed-offsets-in-offset-order

(cherry picked from commit d1853f791147d66303c2700a8e58af762518188c)
Signed-off-by: Jason Gustafson <jason@confluent.io>


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/2b326faf
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/2b326faf
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/2b326faf

Branch: refs/heads/0.11.0
Commit: 2b326faf5f5a2bc632b842c26d2f48a07fa8a15e
Parents: 3321684
Author: Apurva Mehta <apurva@confluent.io>
Authored: Tue May 23 09:45:31 2017 -0700
Committer: Jason Gustafson <jason@confluent.io>
Committed: Tue May 23 09:47:03 2017 -0700

----------------------------------------------------------------------
 .../kafka/coordinator/group/GroupMetadata.scala |  96 ++++++++++++----
 .../group/GroupMetadataManager.scala            |  50 +++++----
 .../group/GroupMetadataManagerTest.scala        | 102 ++++++++++++++++-
 .../coordinator/group/GroupMetadataTest.scala   | 109 ++++++++++++++++++-
 4 files changed, 310 insertions(+), 47 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/2b326faf/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
index 2f76d63..302fcb5 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
@@ -127,6 +127,16 @@ case class GroupSummary(state: String,
                         members: List[MemberSummary])
 
 /**
+  * We cache offset commits along with their commit record offset. This enables us to ensure
that the latest offset
+  * commit is always materialized when we have a mix of transactional and regular offset
commits. Without preserving
+  * information of the commit record offset, compaction of the offsets topic it self may
result in the wrong offset commit
+  * being materialized.
+  */
+case class CommitRecordMetadataAndOffset(appendedBatchOffset: Option[Long], offsetAndMetadata:
OffsetAndMetadata) {
+  def olderThan(that: CommitRecordMetadataAndOffset) : Boolean = appendedBatchOffset.get
< that.appendedBatchOffset.get
+}
+
+/**
  * Group contains the following metadata:
  *
  *  Membership metadata:
@@ -143,12 +153,17 @@ case class GroupSummary(state: String,
 private[group] class GroupMetadata(val groupId: String, initialState: GroupState = Empty)
extends Logging {
 
   private var state: GroupState = initialState
+
   private val members = new mutable.HashMap[String, MemberMetadata]
-  private val offsets = new mutable.HashMap[TopicPartition, OffsetAndMetadata]
+
+  private val offsets = new mutable.HashMap[TopicPartition, CommitRecordMetadataAndOffset]
+
   private val pendingOffsetCommits = new mutable.HashMap[TopicPartition, OffsetAndMetadata]
-  // A map from a producer id to the open offset commits for that producer id.
-  private val pendingTransactionalOffsetCommits = new mutable.HashMap[Long, mutable.Map[TopicPartition,
OffsetAndMetadata]]()
+
+  private val pendingTransactionalOffsetCommits = new mutable.HashMap[Long, mutable.Map[TopicPartition,
CommitRecordMetadataAndOffset]]()
+
   private var receivedTransactionalOffsetCommits = false
+
   private var receivedConsumerOffsetCommits = false
 
   var protocolType: Option[String] = None
@@ -271,19 +286,27 @@ private[group] class GroupMetadata(val groupId: String, initialState:
GroupState
     GroupOverview(groupId, protocolType.getOrElse(""))
   }
 
-  def initializeOffsets(offsets: collection.Map[TopicPartition, OffsetAndMetadata],
-                        pendingTxnOffsets: Map[Long, mutable.Map[TopicPartition, OffsetAndMetadata]])
{
+  def initializeOffsets(offsets: collection.Map[TopicPartition, CommitRecordMetadataAndOffset],
+                        pendingTxnOffsets: Map[Long, mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]])
{
     this.offsets ++= offsets
     this.pendingTransactionalOffsetCommits ++= pendingTxnOffsets
   }
 
-  def completePendingOffsetWrite(topicPartition: TopicPartition, offset: OffsetAndMetadata)
{
-    if (pendingOffsetCommits.contains(topicPartition))
-      offsets.put(topicPartition, offset)
+  def onOffsetCommitAppend(topicPartition: TopicPartition, offsetWithCommitRecordMetadata:
CommitRecordMetadataAndOffset) {
+    if (pendingOffsetCommits.contains(topicPartition)) {
+      if (offsetWithCommitRecordMetadata.appendedBatchOffset.isEmpty)
+        throw new IllegalStateException("Cannot complete offset commit write without providing
the metadata of the record " +
+          "in the log.")
+      if (!offsets.contains(topicPartition) || offsets(topicPartition).olderThan(offsetWithCommitRecordMetadata))
+        offsets.put(topicPartition, offsetWithCommitRecordMetadata)
+    }
 
     pendingOffsetCommits.get(topicPartition) match {
-      case Some(stagedOffset) if offset == stagedOffset => pendingOffsetCommits.remove(topicPartition)
+      case Some(stagedOffset) if offsetWithCommitRecordMetadata.offsetAndMetadata == stagedOffset
=>
+        pendingOffsetCommits.remove(topicPartition)
       case _ =>
+        // The pendingOffsetCommits for this partition could be empty if the topic was deleted,
in which case
+        // its entries would be removed from the cache by the `removeOffsets` method.
     }
   }
 
@@ -301,8 +324,12 @@ private[group] class GroupMetadata(val groupId: String, initialState:
GroupState
 
   def prepareTxnOffsetCommit(producerId: Long, offsets: Map[TopicPartition, OffsetAndMetadata])
{
     receivedTransactionalOffsetCommits = true
-    val producerOffsets = pendingTransactionalOffsetCommits.getOrElseUpdate(producerId, mutable.Map.empty[TopicPartition,
OffsetAndMetadata])
-    producerOffsets ++= offsets
+    val producerOffsets = pendingTransactionalOffsetCommits.getOrElseUpdate(producerId,
+      mutable.Map.empty[TopicPartition, CommitRecordMetadataAndOffset])
+
+    offsets.foreach { case (topicPartition, offsetAndMetadata) =>
+      producerOffsets.put(topicPartition, CommitRecordMetadataAndOffset(None, offsetAndMetadata))
+    }
   }
 
   def hasReceivedConsistentOffsetCommits : Boolean = {
@@ -323,14 +350,32 @@ private[group] class GroupMetadata(val groupId: String, initialState:
GroupState
     }
   }
 
+  def onTxnOffsetCommitAppend(producerId: Long, topicPartition: TopicPartition,
+                              commitRecordMetadataAndOffset: CommitRecordMetadataAndOffset)
{
+    pendingTransactionalOffsetCommits.get(producerId) match {
+      case Some(pendingOffset) =>
+        if (pendingOffset.contains(topicPartition)
+          && pendingOffset(topicPartition).offsetAndMetadata == commitRecordMetadataAndOffset.offsetAndMetadata)
+          pendingOffset.update(topicPartition, commitRecordMetadataAndOffset)
+      case _ =>
+        // We may hit this case if the partition in question has emigrated.
+    }
+  }
+
   /* Complete a pending transactional offset commit. This is called after a commit or abort
marker is fully written
    * to the log.
    */
   def completePendingTxnOffsetCommit(producerId: Long, isCommit: Boolean): Unit = {
     trace(s"Completing transactional offset commit for producer $producerId and group $groupId.
isCommit: $isCommit")
     if (isCommit) {
-      val producerOffsets = pendingTransactionalOffsetCommits.getOrElse(producerId, Map.empty[TopicPartition,
OffsetAndMetadata])
-      offsets ++= producerOffsets
+      val producerOffsets = pendingTransactionalOffsetCommits.getOrElse(producerId, Map.empty[TopicPartition,
CommitRecordMetadataAndOffset])
+      producerOffsets.foreach { case (topicPartition, commitRecordMetadataAndOffset) =>
+        if (commitRecordMetadataAndOffset.appendedBatchOffset.isEmpty)
+          throw new IllegalStateException(s"Trying to complete a transactional offset commit
for producerId $producerId " +
+            s"and groupId $groupId even though the the offset commit record itself hasn't
been appended to the log.")
+        if (!offsets.contains(topicPartition) || offsets(topicPartition).olderThan(commitRecordMetadataAndOffset))
+          offsets.put(topicPartition, commitRecordMetadataAndOffset)
+      }
     }
     pendingTransactionalOffsetCommits.remove(producerId)
   }
@@ -347,21 +392,32 @@ private[group] class GroupMetadata(val groupId: String, initialState:
GroupState
         pendingOffsets.remove(topicPartition)
       }
       val removedOffset = offsets.remove(topicPartition)
-      removedOffset.map(topicPartition -> _)
+      removedOffset.map(topicPartition -> _.offsetAndMetadata)
     }.toMap
   }
 
-  def removeExpiredOffsets(startMs: Long) = {
-    val expiredOffsets = offsets.filter {
-      case (topicPartition, offset) => offset.expireTimestamp < startMs &&
!pendingOffsetCommits.contains(topicPartition)
-    }
+  def removeExpiredOffsets(startMs: Long) : Map[TopicPartition, OffsetAndMetadata] = {
+    val expiredOffsets = offsets
+      .filter {
+        case (topicPartition, commitRecordMetadataAndOffset) =>
+          commitRecordMetadataAndOffset.offsetAndMetadata.expireTimestamp < startMs &&
!pendingOffsetCommits.contains(topicPartition)
+      }
+      .map {
+        case (topicPartition, commitRecordOffsetAndMetadata) =>
+          (topicPartition, commitRecordOffsetAndMetadata.offsetAndMetadata)
+      }
     offsets --= expiredOffsets.keySet
     expiredOffsets.toMap
   }
 
-  def allOffsets = offsets.toMap
+  def allOffsets = offsets.map { case (topicPartition, commitRecordMetadataAndOffset) =>
+    (topicPartition, commitRecordMetadataAndOffset.offsetAndMetadata)
+  }.toMap
+
+  def offset(topicPartition: TopicPartition) : Option[OffsetAndMetadata] = offsets.get(topicPartition).map(_.offsetAndMetadata)
 
-  def offset(topicPartition: TopicPartition) = offsets.get(topicPartition)
+  // visible for testing
+  private[group] def offsetWithRecordMetadata(topicPartition: TopicPartition): Option[CommitRecordMetadataAndOffset]
= offsets.get(topicPartition)
 
   def numOffsets = offsets.size
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/2b326faf/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
index 74a3f7b..f8f536e 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
@@ -309,9 +309,12 @@ class GroupMetadataManager(brokerId: Int,
 
             val responseError = group synchronized {
               if (status.error == Errors.NONE) {
-                if (!group.is(Dead) && !isTxnOffsetCommit) {
+                if (!group.is(Dead)) {
                   filteredOffsetMetadata.foreach { case (topicPartition, offsetAndMetadata)
=>
-                    group.completePendingOffsetWrite(topicPartition, offsetAndMetadata)
+                    if (isTxnOffsetCommit)
+                      group.onTxnOffsetCommitAppend(producerId, topicPartition, CommitRecordMetadataAndOffset(Some(status.baseOffset),
offsetAndMetadata))
+                    else
+                      group.onOffsetCommitAppend(topicPartition, CommitRecordMetadataAndOffset(Some(status.baseOffset),
offsetAndMetadata))
                   }
                 }
                 Errors.NONE
@@ -473,8 +476,8 @@ class GroupMetadataManager(brokerId: Int,
         lazy val buffer = ByteBuffer.allocate(config.loadBufferSize)
 
         // loop breaks if leader changes at any time during the load, since getHighWatermark
is -1
-        val loadedOffsets = mutable.Map[GroupTopicPartition, OffsetAndMetadata]()
-        val pendingOffsets = mutable.Map[Long, mutable.Map[GroupTopicPartition, OffsetAndMetadata]]()
+        val loadedOffsets = mutable.Map[GroupTopicPartition, CommitRecordMetadataAndOffset]()
+        val pendingOffsets = mutable.Map[Long, mutable.Map[GroupTopicPartition, CommitRecordMetadataAndOffset]]()
         val loadedGroups = mutable.Map[String, GroupMetadata]()
         val removedGroups = mutable.Set[String]()
 
@@ -496,10 +499,11 @@ class GroupMetadataManager(brokerId: Int,
               val record = batch.iterator.next()
               val controlRecord = ControlRecordType.parse(record.key)
               if (controlRecord == ControlRecordType.COMMIT) {
-                pendingOffsets.getOrElse(batch.producerId, mutable.Map[GroupTopicPartition,
OffsetAndMetadata]())
+                pendingOffsets.getOrElse(batch.producerId, mutable.Map[GroupTopicPartition,
CommitRecordMetadataAndOffset]())
                   .foreach {
-                    case (groupTopicPartition, offsetAndMetadata) =>
-                      loadedOffsets.put(groupTopicPartition, offsetAndMetadata)
+                    case (groupTopicPartition, commitRecordMetadataAndOffset) =>
+                      if (!loadedOffsets.contains(groupTopicPartition) || loadedOffsets(groupTopicPartition).olderThan(commitRecordMetadataAndOffset))
+                        loadedOffsets.put(groupTopicPartition, commitRecordMetadataAndOffset)
                   }
               }
               pendingOffsets.remove(batch.producerId)
@@ -510,7 +514,7 @@ class GroupMetadataManager(brokerId: Int,
 
                   case offsetKey: OffsetKey =>
                     if (isTxnOffsetCommit && !pendingOffsets.contains(batch.producerId))
-                      pendingOffsets.put(batch.producerId, mutable.Map[GroupTopicPartition,
OffsetAndMetadata]())
+                      pendingOffsets.put(batch.producerId, mutable.Map[GroupTopicPartition,
CommitRecordMetadataAndOffset]())
 
                     // load offset
                     val groupTopicPartition = offsetKey.key
@@ -520,11 +524,11 @@ class GroupMetadataManager(brokerId: Int,
                       else
                         loadedOffsets.remove(groupTopicPartition)
                     } else {
-                      val value = GroupMetadataManager.readOffsetMessageValue(record.value)
+                      val offsetAndMetadata = GroupMetadataManager.readOffsetMessageValue(record.value)
                       if (isTxnOffsetCommit)
-                        pendingOffsets(batch.producerId).put(groupTopicPartition, value)
+                        pendingOffsets(batch.producerId).put(groupTopicPartition, CommitRecordMetadataAndOffset(Some(batch.baseOffset),
offsetAndMetadata))
                       else
-                        loadedOffsets.put(groupTopicPartition, value)
+                        loadedOffsets.put(groupTopicPartition, CommitRecordMetadataAndOffset(Some(batch.baseOffset),
offsetAndMetadata))
                     }
 
                   case groupMetadataKey: GroupMetadataKey =>
@@ -554,15 +558,15 @@ class GroupMetadataManager(brokerId: Int,
             .mapValues(_.map { case (groupTopicPartition, offset) => (groupTopicPartition.topicPartition,
offset) })
             .partition { case (group, _) => loadedGroups.contains(group) }
 
-          val pendingOffsetsByGroup = mutable.Map[String, mutable.Map[Long, mutable.Map[TopicPartition,
OffsetAndMetadata]]]()
+          val pendingOffsetsByGroup = mutable.Map[String, mutable.Map[Long, mutable.Map[TopicPartition,
CommitRecordMetadataAndOffset]]]()
           pendingOffsets.foreach { case (producerId, producerOffsets) =>
             producerOffsets.keySet.map(_.group).foreach(addProducerGroup(producerId, _))
             producerOffsets
               .groupBy(_._1.group)
               .mapValues(_.map { case (groupTopicPartition, offset) => (groupTopicPartition.topicPartition,
offset)})
               .foreach { case (group, offsets) =>
-                val groupPendingOffsets = pendingOffsetsByGroup.getOrElseUpdate(group, mutable.Map.empty[Long,
mutable.Map[TopicPartition, OffsetAndMetadata]])
-                val groupProducerOffsets = groupPendingOffsets.getOrElseUpdate(producerId,
mutable.Map.empty[TopicPartition, OffsetAndMetadata])
+                val groupPendingOffsets = pendingOffsetsByGroup.getOrElseUpdate(group, mutable.Map.empty[Long,
mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]])
+                val groupProducerOffsets = groupPendingOffsets.getOrElseUpdate(producerId,
mutable.Map.empty[TopicPartition, CommitRecordMetadataAndOffset])
                 groupProducerOffsets ++= offsets
               }
           }
@@ -571,8 +575,8 @@ class GroupMetadataManager(brokerId: Int,
             .partition { case (group, _) => loadedGroups.contains(group)}
 
           loadedGroups.values.foreach { group =>
-            val offsets = groupOffsets.getOrElse(group.groupId, Map.empty[TopicPartition,
OffsetAndMetadata])
-            val pendingOffsets = pendingGroupOffsets.getOrElse(group.groupId, Map.empty[Long,
mutable.Map[TopicPartition, OffsetAndMetadata]])
+            val offsets = groupOffsets.getOrElse(group.groupId, Map.empty[TopicPartition,
CommitRecordMetadataAndOffset])
+            val pendingOffsets = pendingGroupOffsets.getOrElse(group.groupId, Map.empty[Long,
mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]])
             loadGroup(group, offsets, pendingOffsets)
             onGroupLoaded(group)
           }
@@ -581,8 +585,8 @@ class GroupMetadataManager(brokerId: Int,
           // metadata stored in the log
           (emptyGroupOffsets.keySet ++ pendingEmptyGroupOffsets.keySet).foreach { case(groupId)
=>
             val group = new GroupMetadata(groupId)
-            val offsets = emptyGroupOffsets.getOrElse(groupId, Map.empty[TopicPartition,
OffsetAndMetadata])
-            val pendingOffsets = pendingEmptyGroupOffsets.getOrElse(groupId, Map.empty[Long,
mutable.Map[TopicPartition, OffsetAndMetadata]])
+            val offsets = emptyGroupOffsets.getOrElse(groupId, Map.empty[TopicPartition,
CommitRecordMetadataAndOffset])
+            val pendingOffsets = pendingEmptyGroupOffsets.getOrElse(groupId, Map.empty[Long,
mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]])
             loadGroup(group, offsets, pendingOffsets)
             onGroupLoaded(group)
           }
@@ -603,17 +607,19 @@ class GroupMetadataManager(brokerId: Int,
     }
   }
 
-  private def loadGroup(group: GroupMetadata, offsets: Map[TopicPartition, OffsetAndMetadata],
-                        pendingTransactionalOffsets: Map[Long, mutable.Map[TopicPartition,
OffsetAndMetadata]]): Unit = {
+  private def loadGroup(group: GroupMetadata, offsets: Map[TopicPartition, CommitRecordMetadataAndOffset],
+                        pendingTransactionalOffsets: Map[Long, mutable.Map[TopicPartition,
CommitRecordMetadataAndOffset]]): Unit = {
     // offsets are initialized prior to loading the group into the cache to ensure that clients
see a consistent
     // view of the group's offsets
-    val loadedOffsets = offsets.mapValues { offsetAndMetadata =>
+    val loadedOffsets = offsets.mapValues { case CommitRecordMetadataAndOffset(commitRecordOffset,
offsetAndMetadata) =>
       // special handling for version 0:
       // set the expiration time stamp as commit time stamp + server default retention time
-      if (offsetAndMetadata.expireTimestamp == org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP)
+      val updatedOffsetAndMetadata =
+        if (offsetAndMetadata.expireTimestamp == org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP)
         offsetAndMetadata.copy(expireTimestamp = offsetAndMetadata.commitTimestamp + config.offsetsRetentionMs)
       else
         offsetAndMetadata
+      CommitRecordMetadataAndOffset(commitRecordOffset, updatedOffsetAndMetadata)
     }
     trace(s"Initialized offsets $loadedOffsets for group ${group.groupId}")
     group.initializeOffsets(loadedOffsets, pendingTransactionalOffsets.toMap)

http://git-wip-us.apache.org/repos/asf/kafka/blob/2b326faf/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
index f76eb7b..8318741 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
@@ -278,6 +278,7 @@ class GroupMetadataManagerTest {
 
     val buffer = ByteBuffer.allocate(1024)
     var nextOffset = 0
+    val commitOffsetsLogPosition = nextOffset
     nextOffset += appendTransactionalOffsetCommits(buffer, producerId, producerEpoch, nextOffset,
committedOffsets)
     nextOffset += completeTransactionalOffsetCommit(buffer, producerId, producerEpoch, nextOffset,
isCommit = true)
     nextOffset += appendTransactionalOffsetCommits(buffer, producerId, producerEpoch, nextOffset,
abortedOffsets)
@@ -301,6 +302,7 @@ class GroupMetadataManagerTest {
     assertEquals(committedOffsets.size, group.allOffsets.size)
     committedOffsets.foreach { case (topicPartition, offset) =>
       assertEquals(Some(offset), group.offset(topicPartition).map(_.offset))
+      assertEquals(Some(commitOffsetsLogPosition), group.offsetWithRecordMetadata(topicPartition).head.appendedBatchOffset)
     }
 
     // We should have pending commits.
@@ -335,9 +337,13 @@ class GroupMetadataManagerTest {
     )
 
     val buffer = ByteBuffer.allocate(1024)
-    var nextOffset = 0
+    var nextOffset = 0L
+
+    val firstProduceRecordOffset = nextOffset
     nextOffset += appendTransactionalOffsetCommits(buffer, firstProducerId, firstProducerEpoch,
nextOffset, committedOffsetsFirstProducer)
     nextOffset += completeTransactionalOffsetCommit(buffer, firstProducerId, firstProducerEpoch,
nextOffset, isCommit = true)
+
+    val secondProducerRecordOffset = nextOffset
     nextOffset += appendTransactionalOffsetCommits(buffer, secondProducerId, secondProducerEpoch,
nextOffset, committedOffsetsSecondProducer)
     nextOffset += completeTransactionalOffsetCommit(buffer, secondProducerId, secondProducerEpoch,
nextOffset, isCommit = true)
     buffer.flip()
@@ -358,12 +364,106 @@ class GroupMetadataManagerTest {
     assertEquals(committedOffsetsFirstProducer.size + committedOffsetsSecondProducer.size,
group.allOffsets.size)
     committedOffsetsFirstProducer.foreach { case (topicPartition, offset) =>
       assertEquals(Some(offset), group.offset(topicPartition).map(_.offset))
+      assertEquals(Some(firstProduceRecordOffset), group.offsetWithRecordMetadata(topicPartition).head.appendedBatchOffset)
     }
     committedOffsetsSecondProducer.foreach { case (topicPartition, offset) =>
       assertEquals(Some(offset), group.offset(topicPartition).map(_.offset))
+      assertEquals(Some(secondProducerRecordOffset), group.offsetWithRecordMetadata(topicPartition).head.appendedBatchOffset)
+    }
+  }
+
+  @Test
+  def testGroupLoadWithConsumerAndTransactionalOffsetCommitsConsumerWins(): Unit = {
+    val groupMetadataTopicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME,
groupPartitionId)
+    val producerId = 1000L
+    val producerEpoch: Short = 2
+
+    val transactionalOffsetCommits = Map(
+      new TopicPartition("foo", 0) -> 23L
+    )
+
+    val consumerOffsetCommits = Map(
+      new TopicPartition("foo", 0) -> 24L
+    )
+
+    val buffer = ByteBuffer.allocate(1024)
+    var nextOffset = 0
+    nextOffset += appendTransactionalOffsetCommits(buffer, producerId, producerEpoch, nextOffset,
transactionalOffsetCommits)
+    val consumerRecordOffset = nextOffset
+    nextOffset += appendConsumerOffsetCommit(buffer, nextOffset, consumerOffsetCommits)
+    nextOffset += completeTransactionalOffsetCommit(buffer, producerId, producerEpoch, nextOffset,
isCommit = true)
+    buffer.flip()
+
+    val records = MemoryRecords.readableRecords(buffer)
+    expectGroupMetadataLoad(groupMetadataTopicPartition, 0, records)
+
+    EasyMock.replay(replicaManager)
+
+    groupMetadataManager.loadGroupsAndOffsets(groupMetadataTopicPartition, _ => ())
+
+    // The group should be loaded with pending offsets.
+    val group = groupMetadataManager.getGroup(groupId).getOrElse(fail("Group was not loaded
into the cache"))
+    assertEquals(groupId, group.groupId)
+    assertEquals(Empty, group.currentState)
+    assertEquals(1, group.allOffsets.size)
+    assertTrue(group.hasOffsets)
+    assertFalse(group.hasPendingOffsetCommitsFromProducer(producerId))
+    assertEquals(consumerOffsetCommits.size, group.allOffsets.size)
+    consumerOffsetCommits.foreach { case (topicPartition, offset) =>
+      assertEquals(Some(offset), group.offset(topicPartition).map(_.offset))
+      assertEquals(Some(consumerRecordOffset), group.offsetWithRecordMetadata(topicPartition).head.appendedBatchOffset)
     }
   }
 
+  @Test
+  def testGroupLoadWithConsumerAndTransactionalOffsetCommitsTransactionWins(): Unit = {
+    val groupMetadataTopicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME,
groupPartitionId)
+    val producerId = 1000L
+    val producerEpoch: Short = 2
+
+    val transactionalOffsetCommits = Map(
+      new TopicPartition("foo", 0) -> 23L
+    )
+
+    val consumerOffsetCommits = Map(
+      new TopicPartition("foo", 0) -> 24L
+    )
+
+    val buffer = ByteBuffer.allocate(1024)
+    var nextOffset = 0
+    nextOffset += appendConsumerOffsetCommit(buffer, nextOffset, consumerOffsetCommits)
+    nextOffset += appendTransactionalOffsetCommits(buffer, producerId, producerEpoch, nextOffset,
transactionalOffsetCommits)
+    nextOffset += completeTransactionalOffsetCommit(buffer, producerId, producerEpoch, nextOffset,
isCommit = true)
+    buffer.flip()
+
+    val records = MemoryRecords.readableRecords(buffer)
+    expectGroupMetadataLoad(groupMetadataTopicPartition, 0, records)
+
+    EasyMock.replay(replicaManager)
+
+    groupMetadataManager.loadGroupsAndOffsets(groupMetadataTopicPartition, _ => ())
+
+    // The group should be loaded with pending offsets.
+    val group = groupMetadataManager.getGroup(groupId).getOrElse(fail("Group was not loaded
into the cache"))
+    assertEquals(groupId, group.groupId)
+    assertEquals(Empty, group.currentState)
+    assertEquals(1, group.allOffsets.size)
+    assertTrue(group.hasOffsets)
+    assertFalse(group.hasPendingOffsetCommitsFromProducer(producerId))
+    assertEquals(consumerOffsetCommits.size, group.allOffsets.size)
+    transactionalOffsetCommits.foreach { case (topicPartition, offset) =>
+      assertEquals(Some(offset), group.offset(topicPartition).map(_.offset))
+    }
+  }
+
+  private def appendConsumerOffsetCommit(buffer: ByteBuffer, baseOffset: Long, offsets: Map[TopicPartition,
Long]) = {
+    val builder = MemoryRecords.builder(buffer, CompressionType.NONE, TimestampType.LOG_APPEND_TIME,
baseOffset)
+    val commitRecords = createCommittedOffsetRecords(offsets)
+    commitRecords.foreach(builder.append)
+    builder.build()
+    offsets.size
+  }
+
   private def appendTransactionalOffsetCommits(buffer: ByteBuffer, producerId: Long, producerEpoch:
Short,
                                                baseOffset: Long, offsets: Map[TopicPartition,
Long]): Int = {
     val builder = MemoryRecords.builder(buffer, CompressionType.NONE, baseOffset, producerId,
producerEpoch, 0, true)

http://git-wip-us.apache.org/repos/asf/kafka/blob/2b326faf/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataTest.scala
index e62c0d3..0e13f89 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataTest.scala
@@ -296,12 +296,13 @@ class GroupMetadataTest extends JUnitSuite {
   def testOffsetCommit(): Unit = {
     val partition = new TopicPartition("foo", 0)
     val offset = OffsetAndMetadata(37)
+    val commitRecordOffset = 3
 
     group.prepareOffsetCommit(Map(partition -> offset))
     assertTrue(group.hasOffsets)
     assertEquals(None, group.offset(partition))
 
-    group.completePendingOffsetWrite(partition, offset)
+    group.onOffsetCommitAppend(partition, CommitRecordMetadataAndOffset(Some(commitRecordOffset),
offset))
     assertTrue(group.hasOffsets)
     assertEquals(Some(offset), group.offset(partition))
   }
@@ -337,7 +338,7 @@ class GroupMetadataTest extends JUnitSuite {
     assertTrue(group.hasOffsets)
     assertEquals(None, group.offset(partition))
 
-    group.completePendingOffsetWrite(partition, secondOffset)
+    group.onOffsetCommitAppend(partition, CommitRecordMetadataAndOffset(Some(3L), secondOffset))
     assertTrue(group.hasOffsets)
     assertEquals(Some(secondOffset), group.offset(partition))
   }
@@ -355,15 +356,115 @@ class GroupMetadataTest extends JUnitSuite {
     group.prepareOffsetCommit(Map(partition -> secondOffset))
     assertTrue(group.hasOffsets)
 
-    group.completePendingOffsetWrite(partition, firstOffset)
+    group.onOffsetCommitAppend(partition, CommitRecordMetadataAndOffset(Some(4L), firstOffset))
     assertTrue(group.hasOffsets)
     assertEquals(Some(firstOffset), group.offset(partition))
 
-    group.completePendingOffsetWrite(partition, secondOffset)
+    group.onOffsetCommitAppend(partition, CommitRecordMetadataAndOffset(Some(5L), secondOffset))
     assertTrue(group.hasOffsets)
     assertEquals(Some(secondOffset), group.offset(partition))
   }
 
+  @Test
+  def testConsumerBeatsTransactionalOffsetCommit(): Unit = {
+    val partition = new TopicPartition("foo", 0)
+    val producerId = 13232L
+    val txnOffsetCommit = OffsetAndMetadata(37)
+    val consumerOffsetCommit = OffsetAndMetadata(57)
+
+    group.prepareTxnOffsetCommit(producerId, Map(partition -> txnOffsetCommit))
+    assertTrue(group.hasOffsets)
+    assertEquals(None, group.offset(partition))
+
+    group.prepareOffsetCommit(Map(partition -> consumerOffsetCommit))
+    assertTrue(group.hasOffsets)
+
+    group.onTxnOffsetCommitAppend(producerId, partition, CommitRecordMetadataAndOffset(Some(3L),
txnOffsetCommit))
+    group.onOffsetCommitAppend(partition, CommitRecordMetadataAndOffset(Some(4L), consumerOffsetCommit))
+    assertTrue(group.hasOffsets)
+    assertEquals(Some(consumerOffsetCommit), group.offset(partition))
+
+    group.completePendingTxnOffsetCommit(producerId, isCommit = true)
+    assertTrue(group.hasOffsets)
+    // This is the crucial assertion which validates that we materialize offsets in offset
order, not transactional order.
+    assertEquals(Some(consumerOffsetCommit), group.offset(partition))
+  }
+
+  @Test
+  def testTransactionBeatsConsumerOffsetCommit(): Unit = {
+    val partition = new TopicPartition("foo", 0)
+    val producerId = 13232L
+    val txnOffsetCommit = OffsetAndMetadata(37)
+    val consumerOffsetCommit = OffsetAndMetadata(57)
+
+    group.prepareTxnOffsetCommit(producerId, Map(partition -> txnOffsetCommit))
+    assertTrue(group.hasOffsets)
+    assertEquals(None, group.offset(partition))
+
+    group.prepareOffsetCommit(Map(partition -> consumerOffsetCommit))
+    assertTrue(group.hasOffsets)
+
+    group.onOffsetCommitAppend(partition, CommitRecordMetadataAndOffset(Some(3L), consumerOffsetCommit))
+    group.onTxnOffsetCommitAppend(producerId, partition, CommitRecordMetadataAndOffset(Some(4L),
txnOffsetCommit))
+    assertTrue(group.hasOffsets)
+    // The transactional offset commit hasn't been committed yet, so we should materialize
the consumer offset commit.
+    assertEquals(Some(consumerOffsetCommit), group.offset(partition))
+
+    group.completePendingTxnOffsetCommit(producerId, isCommit = true)
+    assertTrue(group.hasOffsets)
+    // The transactional offset commit has been materialized and the transactional commit
record is later in the log,
+    // so it should be materialized.
+    assertEquals(Some(txnOffsetCommit), group.offset(partition))
+  }
+
+  @Test
+  def testTransactionalCommitIsAbortedAndConsumerCommitWins(): Unit = {
+    val partition = new TopicPartition("foo", 0)
+    val producerId = 13232L
+    val txnOffsetCommit = OffsetAndMetadata(37)
+    val consumerOffsetCommit = OffsetAndMetadata(57)
+
+    group.prepareTxnOffsetCommit(producerId, Map(partition -> txnOffsetCommit))
+    assertTrue(group.hasOffsets)
+    assertEquals(None, group.offset(partition))
+
+    group.prepareOffsetCommit(Map(partition -> consumerOffsetCommit))
+    assertTrue(group.hasOffsets)
+
+    group.onOffsetCommitAppend(partition, CommitRecordMetadataAndOffset(Some(3L), consumerOffsetCommit))
+    group.onTxnOffsetCommitAppend(producerId, partition, CommitRecordMetadataAndOffset(Some(4L),
txnOffsetCommit))
+    assertTrue(group.hasOffsets)
+    // The transactional offset commit hasn't been committed yet, so we should materialize
the consumer offset commit.
+    assertEquals(Some(consumerOffsetCommit), group.offset(partition))
+
+    group.completePendingTxnOffsetCommit(producerId, isCommit = false)
+    assertTrue(group.hasOffsets)
+    // The transactional offset commit should be discarded and the consumer offset commit
should continue to be
+    // materialized.
+    assertFalse(group.hasPendingOffsetCommitsFromProducer(producerId))
+    assertEquals(Some(consumerOffsetCommit), group.offset(partition))
+  }
+
+  @Test
+  def testFailedTxnOffsetCommitLeavesNoPendingState(): Unit = {
+    val partition = new TopicPartition("foo", 0)
+    val producerId = 13232L
+    val txnOffsetCommit = OffsetAndMetadata(37)
+
+    group.prepareTxnOffsetCommit(producerId, Map(partition -> txnOffsetCommit))
+    assertTrue(group.hasPendingOffsetCommitsFromProducer(producerId))
+    assertTrue(group.hasOffsets)
+    assertEquals(None, group.offset(partition))
+    group.failPendingTxnOffsetCommit(producerId, partition, txnOffsetCommit)
+    assertFalse(group.hasOffsets)
+    assertFalse(group.hasPendingOffsetCommitsFromProducer(producerId))
+
+    // The commit marker should now have no effect.
+    group.completePendingTxnOffsetCommit(producerId, isCommit = true)
+    assertFalse(group.hasOffsets)
+    assertFalse(group.hasPendingOffsetCommitsFromProducer(producerId))
+  }
+
   private def assertState(group: GroupMetadata, targetState: GroupState) {
     val states: Set[GroupState] = Set(Stable, PreparingRebalance, AwaitingSync, Dead)
     val otherStates = states - targetState


Mime
View raw message