brandboat commented on code in PR #17914:
URL: https://github.com/apache/kafka/pull/17914#discussion_r1864078328
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -194,6 +209,104 @@ class GroupCoordinatorBaseRequestTest(cluster:
ClusterInstance) {
assertEquals(expectedResponse, response.data)
}
+ protected def commitTxnOffset(
+ groupId: String,
+ memberId: String,
+ generationId: Int,
+ producerId: Long,
+ producerEpoch: Short,
+ transactionalId: String,
+ topic: String,
+ partition: Int,
+ offset: Long,
+ expectedError: Errors,
+ version: Short =
ApiKeys.TXN_OFFSET_COMMIT.latestVersion(isUnstableApiEnabled)
+ ): Unit = {
+ val request = new TxnOffsetCommitRequest.Builder(
+ new TxnOffsetCommitRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setGenerationId(generationId)
+ .setProducerId(producerId)
+ .setProducerEpoch(producerEpoch)
+ .setTransactionalId(transactionalId)
+ .setTopics(List(
+ new TxnOffsetCommitRequestData.TxnOffsetCommitRequestTopic()
+ .setName(topic)
+ .setPartitions(List(
+ new TxnOffsetCommitRequestData.TxnOffsetCommitRequestPartition()
+ .setPartitionIndex(partition)
+ .setCommittedOffset(offset)
+ ).asJava)
+ ).asJava)
+ ).build(version)
+
+ val expectedResponse = new TxnOffsetCommitResponseData()
+ .setTopics(List(
+ new TxnOffsetCommitResponseData.TxnOffsetCommitResponseTopic()
+ .setName(topic)
+ .setPartitions(List(
+ new TxnOffsetCommitResponseData.TxnOffsetCommitResponsePartition()
+ .setPartitionIndex(partition)
+ .setErrorCode(expectedError.code)
+ ).asJava)
+ ).asJava)
+
+ val response = connectAndReceive[TxnOffsetCommitResponse](request)
+ assertEquals(expectedResponse, response.data)
+ }
+
+ protected def addOffsetsToTxn(
+ groupId: String,
+ producerId: Long,
+ producerEpoch: Short,
+ transactionalId: String,
+ version: Short =
ApiKeys.ADD_OFFSETS_TO_TXN.latestVersion(isUnstableApiEnabled)
+ ): Unit = {
+ val request = new AddOffsetsToTxnRequest.Builder(
+ new AddOffsetsToTxnRequestData()
+ .setTransactionalId(transactionalId)
+ .setProducerId(producerId)
+ .setProducerEpoch(producerEpoch)
+ .setGroupId(groupId)
+ ).build(version)
+
+ val response = connectAndReceive[AddOffsetsToTxnResponse](request)
+ assertEquals(new AddOffsetsToTxnResponseData(), response.data)
+ }
+
+ protected def initProducerId(
+ transactionalId: String,
+ transactionTimeoutMs: Int = 60000,
+ producerIdAndEpoch: ProducerIdAndEpoch,
+ expectedError: Errors,
+ version: Short =
ApiKeys.INIT_PRODUCER_ID.latestVersion(isUnstableApiEnabled)
+ ): ProducerIdAndEpoch = {
+ val request = new InitProducerIdRequest.Builder(
+ new InitProducerIdRequestData()
+ .setTransactionalId(transactionalId)
+ .setTransactionTimeoutMs(transactionTimeoutMs)
+ .setProducerId(producerIdAndEpoch.producerId)
+ .setProducerEpoch(producerIdAndEpoch.epoch))
+ .build(version)
+
+ val response = connectAndReceive[InitProducerIdResponse](request).data()
+ assertEquals(expectedError.code(), response.errorCode())
+ new ProducerIdAndEpoch(response.producerId(), response.producerEpoch())
+ }
+
+ protected def writeTxnMarkers(
Review Comment:
In the beginning I just want to put pending transaction to complete thus I
can get txn offset through `OFFSET_FETCH` RPC, abortTransaction and
endTransaction both can make this happen, but you are right, call `END_TXN` RPC
here is more suitable here. Thanks for the suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
