Add OffsetFetch to group concurrency test
As title suggested, consumers would first do an OffsetFetch before starting the normal processing. It makes sense to add it to the concurrent test suite to verify whether there would be a blocking behavior.
Reviewers: Guozhang Wang
@@ -29,7 +29,7 @@ import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.message.LeaveGroupRequestData.MemberIdentity
import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.JoinGroupRequest
+import org.apache.kafka.common.requests.{JoinGroupRequest, OffsetFetchResponse}
import org.apache.kafka.common.utils.Time
import org.easymock.EasyMock
import org.junit.Assert._
@@ -53,13 +53,16 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest
private val allOperations = Seq(
new JoinGroupOperation,
new SyncGroupOperation,
+ new OffsetFetchOperation,
new CommitOffsetsOperation,
new HeartbeatOperation,
new LeaveGroupOperation
)
+
private val allOperationsWithTxn = Seq(
new JoinGroupOperation,
new SyncGroupOperation,
+ new OffsetFetchOperation,
new CommitTxnOffsetsOperation,
new CompleteTxnOperation,
new HeartbeatOperation,
@@ -176,7 +179,7 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest
class SyncGroupOperation extends GroupOperation[SyncGroupCallbackParams, SyncGroupCallback] {
override def responseCallback(responsePromise: Promise[SyncGroupCallbackParams]): SyncGroupCallback = {
val callback: SyncGroupCallback = syncGroupResult =>
- responsePromise.success(syncGroupResult.memberAssignment, syncGroupResult.error)
+ responsePromise.success(syncGroupResult.error, syncGroupResult.memberAssignment)
callback
}
override def runWithCallback(member: GroupMember, responseCallback: SyncGroupCallback): Unit = {
@@ -190,7 +193,9 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest
}
override def awaitAndVerify(member: GroupMember): Unit = {
val result = await(member, DefaultSessionTimeout)
- assertEquals(Errors.NONE, result._2)
+ assertEquals(Errors.NONE, result._1)
+ assertNotNull(result._2)
+ assertEquals(0, result._2.length)
}
}
@@ -209,6 +214,22 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest
}
}
+ class OffsetFetchOperation extends GroupOperation[OffsetFetchCallbackParams, OffsetFetchCallback] {
+ override def responseCallback(responsePromise: Promise[OffsetFetchCallbackParams]): OffsetFetchCallback = {
+ val callback: OffsetFetchCallback = (error, offsets) => responsePromise.success(error, offsets)
+ callback
+ }
+ override def runWithCallback(member: GroupMember, responseCallback: OffsetFetchCallback): Unit = {
+ val (error, partitionData) = groupCoordinator.handleFetchOffsets(member.groupId, requireStable = true, None)
+ responseCallback(error, partitionData)
+ }
+ override def awaitAndVerify(member: GroupMember): Unit = {
+ val result = await(member, 500)
+ assertEquals(Errors.NONE, result._1)
+ assertEquals(Map.empty, result._2)
+ }
+ }
+
class CommitOffsetsOperation extends GroupOperation[CommitOffsetCallbackParams, CommitOffsetCallback] {
override def responseCallback(responsePromise: Promise[CommitOffsetCallbackParams]): CommitOffsetCallback = {
val callback: CommitOffsetCallback = offsets => responsePromise.success(offsets)
@@ -290,10 +311,12 @@ object GroupCoordinatorConcurrencyTest {
type JoinGroupCallbackParams = JoinGroupResult
type JoinGroupCallback = JoinGroupResult => Unit
- type SyncGroupCallbackParams = (Array[Byte], Errors)
+ type SyncGroupCallbackParams = (Errors, Array[Byte])
type SyncGroupCallback = SyncGroupResult => Unit
type HeartbeatCallbackParams = Errors
type HeartbeatCallback = Errors => Unit
+ type OffsetFetchCallbackParams = (Errors, Map[TopicPartition, OffsetFetchResponse.PartitionData])
+ type OffsetFetchCallback = (Errors, Map[TopicPartition, OffsetFetchResponse.PartitionData]) => Unit
type CommitOffsetCallbackParams = Map[TopicPartition, Errors]
type CommitOffsetCallback = Map[TopicPartition, Errors] => Unit
type LeaveGroupCallbackParams = LeaveGroupResult
