You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2023/12/21 19:03:33 UTC
(kafka) branch 3.7 updated: KAFKA-14505; [3/N] Wire WriteTxnMarkers API (#14985)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch 3.7
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.7 by this push:
new 620fce7a00b KAFKA-14505; [3/N] Wire WriteTxnMarkers API (#14985)
620fce7a00b is described below
commit 620fce7a00bd1d23d5c276f5f5534cb18dd52772
Author: David Jacot <dj...@confluent.io>
AuthorDate: Thu Dec 21 19:59:41 2023 +0100
KAFKA-14505; [3/N] Wire WriteTxnMarkers API (#14985)
This patch wires the handling of makers written by the transaction coordinator via the WriteTxnMarkers API. In the old group coordinator, the markers are written to the logs and the group coordinator is informed to materialize the changes as a second step if the writes were successful. This approach does not really work with the new group coordinator for mainly two reasons: 1) The second step would actually fail while the coordinator is loading and there is no guarantee that the loadi [...]
Reviewers: Jeff Kim <je...@confluent.io>, Justine Olshan <jo...@confluent.io>
---
.../coordinator/group/CoordinatorLoaderImpl.scala | 20 +-
.../group/CoordinatorPartitionWriter.scala | 84 +++--
.../group/GroupCoordinatorAdapter.scala | 14 +
core/src/main/scala/kafka/server/KafkaApis.scala | 102 ++++--
.../integration/kafka/api/TransactionsTest.scala | 45 ++-
.../group/CoordinatorLoaderImplTest.scala | 68 +++-
.../group/CoordinatorPartitionWriterTest.scala | 79 +++-
.../scala/unit/kafka/server/KafkaApisTest.scala | 225 +++++++++++-
.../kafka/coordinator/group/GroupCoordinator.java | 24 ++
.../coordinator/group/GroupCoordinatorService.java | 40 +-
.../coordinator/group/GroupCoordinatorShard.java | 18 +
.../coordinator/group/OffsetMetadataManager.java | 40 ++
.../group/runtime/CoordinatorPlayback.java | 16 +
.../group/runtime/CoordinatorRuntime.java | 197 ++++++++++
.../group/runtime/CoordinatorShard.java | 19 +-
.../coordinator/group/runtime/PartitionWriter.java | 20 +
.../group/runtime/SnapshottableCoordinator.java | 17 +
.../group/GroupCoordinatorServiceTest.java | 88 ++++-
.../group/GroupCoordinatorShardTest.java | 33 ++
.../group/OffsetMetadataManagerTest.java | 77 ++++
.../group/runtime/CoordinatorRuntimeTest.java | 408 ++++++++++++++++++++-
.../group/runtime/InMemoryPartitionWriter.java | 172 ++++++++-
22 files changed, 1718 insertions(+), 88 deletions(-)
diff --git a/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala b/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala
index c583e693224..7666bf963ac 100644
--- a/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala
+++ b/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala
@@ -20,7 +20,8 @@ import kafka.server.ReplicaManager
import kafka.utils.Logging
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.errors.NotLeaderOrFollowerException
-import org.apache.kafka.common.record.{FileRecords, MemoryRecords}
+import org.apache.kafka.common.record.{ControlRecordType, FileRecords, MemoryRecords}
+import org.apache.kafka.common.requests.TransactionResult
import org.apache.kafka.common.utils.Time
import org.apache.kafka.coordinator.group.runtime.CoordinatorLoader.{Deserializer, LoadSummary, UnknownRecordTypeException}
import org.apache.kafka.coordinator.group.runtime.{CoordinatorLoader, CoordinatorPlayback}
@@ -135,7 +136,22 @@ class CoordinatorLoaderImpl[T](
memoryRecords.batches.forEach { batch =>
if (batch.isControlBatch) {
- throw new IllegalStateException("Control batches are not supported yet.")
+ batch.asScala.foreach { record =>
+ val controlRecord = ControlRecordType.parse(record.key)
+ if (controlRecord == ControlRecordType.COMMIT) {
+ coordinator.replayEndTransactionMarker(
+ batch.producerId,
+ batch.producerEpoch,
+ TransactionResult.COMMIT
+ )
+ } else if (controlRecord == ControlRecordType.ABORT) {
+ coordinator.replayEndTransactionMarker(
+ batch.producerId,
+ batch.producerEpoch,
+ TransactionResult.ABORT
+ )
+ }
+ }
} else {
batch.asScala.foreach { record =>
numRecords = numRecords + 1
diff --git a/core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala b/core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala
index 2c12a30e0f0..c8c8625ced4 100644
--- a/core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala
+++ b/core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala
@@ -21,9 +21,10 @@ import kafka.server.{ActionQueue, ReplicaManager}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.errors.RecordTooLargeException
import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch, TimestampType}
+import org.apache.kafka.common.record.{CompressionType, ControlRecordType, EndTransactionMarker, MemoryRecords, RecordBatch, TimestampType}
import org.apache.kafka.common.record.Record.EMPTY_HEADERS
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.requests.TransactionResult
import org.apache.kafka.common.utils.{BufferSupplier, Time}
import org.apache.kafka.coordinator.group.runtime.PartitionWriter
import org.apache.kafka.storage.internals.log.AppendOrigin
@@ -160,28 +161,7 @@ class CoordinatorPartitionWriter[T](
s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
}
- var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
- replicaManager.appendRecords(
- timeout = 0L,
- requiredAcks = 1,
- internalTopicsAllowed = true,
- origin = AppendOrigin.COORDINATOR,
- entriesPerPartition = Map(tp -> recordsBuilder.build()),
- responseCallback = results => appendResults = results,
- // We can directly complete the purgatories here because we don't hold
- // any conflicting locks.
- actionQueue = directActionQueue
- )
-
- val partitionResult = appendResults.getOrElse(tp,
- throw new IllegalStateException(s"Append status $appendResults should have partition $tp."))
-
- if (partitionResult.error != Errors.NONE) {
- throw partitionResult.error.exception()
- }
-
- // Required offset.
- partitionResult.lastOffset + 1
+ internalAppend(tp, recordsBuilder.build())
} finally {
bufferSupplier.release(buffer)
}
@@ -190,4 +170,62 @@ class CoordinatorPartitionWriter[T](
throw Errors.NOT_LEADER_OR_FOLLOWER.exception()
}
}
+
+ /**
+ * Write the transaction end marker.
+ *
+ * @param tp The partition to write records to.
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param coordinatorEpoch The epoch of the transaction coordinator.
+ * @param result The transaction result.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def appendEndTransactionMarker(
+ tp: TopicPartition,
+ producerId: Long,
+ producerEpoch: Short,
+ coordinatorEpoch: Int,
+ result: TransactionResult
+ ): Long = {
+ val controlRecordType = result match {
+ case TransactionResult.COMMIT => ControlRecordType.COMMIT
+ case TransactionResult.ABORT => ControlRecordType.ABORT
+ }
+
+ internalAppend(tp, MemoryRecords.withEndTransactionMarker(
+ producerId,
+ producerEpoch,
+ new EndTransactionMarker(controlRecordType, coordinatorEpoch)
+ ))
+ }
+
+ private def internalAppend(
+ tp: TopicPartition,
+ memoryRecords: MemoryRecords
+ ): Long = {
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
+ internalTopicsAllowed = true,
+ origin = AppendOrigin.COORDINATOR,
+ entriesPerPartition = Map(tp -> memoryRecords),
+ responseCallback = results => appendResults = results,
+ // We can directly complete the purgatories here because we don't hold
+ // any conflicting locks.
+ actionQueue = directActionQueue
+ )
+
+ val partitionResult = appendResults.getOrElse(tp,
+ throw new IllegalStateException(s"Append status $appendResults should have partition $tp."))
+
+ if (partitionResult.error != Errors.NONE) {
+ throw partitionResult.error.exception()
+ }
+
+ // Required offset.
+ partitionResult.lastOffset + 1
+ }
}
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
index 678844cc662..67cada04962 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
@@ -29,6 +29,7 @@ import org.apache.kafka.common.utils.{BufferSupplier, Time}
import org.apache.kafka.image.{MetadataDelta, MetadataImage}
import org.apache.kafka.server.util.FutureUtils
+import java.time.Duration
import java.util
import java.util.{Optional, OptionalInt, Properties}
import java.util.concurrent.CompletableFuture
@@ -546,6 +547,19 @@ private[group] class GroupCoordinatorAdapter(
future
}
+ override def completeTransaction(
+ tp: TopicPartition,
+ producerId: Long,
+ producerEpoch: Short,
+ coordinatorEpoch: Int,
+ result: TransactionResult,
+ timeout: Duration
+ ): CompletableFuture[Void] = {
+ FutureUtils.failedFuture(new IllegalStateException(
+ s"The old group coordinator does not support `completeTransaction` API."
+ ))
+ }
+
override def partitionFor(groupId: String): Int = {
coordinator.partitionFor(groupId)
}
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index bb4dce81ef6..872d5262c5c 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -77,6 +77,7 @@ import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchIsolation, Fet
import java.lang.{Long => JLong}
import java.nio.ByteBuffer
+import java.time.Duration
import java.util
import java.util.concurrent.{CompletableFuture, ConcurrentHashMap}
import java.util.concurrent.atomic.AtomicInteger
@@ -2361,25 +2362,27 @@ class KafkaApis(val requestChannel: RequestChannel,
* request, so there could be multiple appends of markers to the log. The final response will be sent only
* after all appends have returned.
*/
- def maybeSendResponseCallback(producerId: Long, result: TransactionResult)(responseStatus: Map[TopicPartition, PartitionResponse]): Unit = {
- trace(s"End transaction marker append for producer id $producerId completed with status: $responseStatus")
- val currentErrors = new ConcurrentHashMap[TopicPartition, Errors](responseStatus.map { case (k, v) => k -> v.error }.asJava)
+ def maybeSendResponseCallback(producerId: Long, result: TransactionResult, currentErrors: ConcurrentHashMap[TopicPartition, Errors]): Unit = {
+ trace(s"End transaction marker append for producer id $producerId completed with status: $currentErrors")
updateErrors(producerId, currentErrors)
- val successfulOffsetsPartitions = responseStatus.filter { case (topicPartition, partitionResponse) =>
- topicPartition.topic == GROUP_METADATA_TOPIC_NAME && partitionResponse.error == Errors.NONE
- }.keys
- if (successfulOffsetsPartitions.nonEmpty) {
- // as soon as the end transaction marker has been written for a transactional offset commit,
- // call to the group coordinator to materialize the offsets into the cache
- try {
- groupCoordinator.onTransactionCompleted(producerId, successfulOffsetsPartitions.asJava, result)
- } catch {
- case e: Exception =>
- error(s"Received an exception while trying to update the offsets cache on transaction marker append", e)
- val updatedErrors = new ConcurrentHashMap[TopicPartition, Errors]()
- successfulOffsetsPartitions.foreach(updatedErrors.put(_, Errors.UNKNOWN_SERVER_ERROR))
- updateErrors(producerId, updatedErrors)
+ if (!config.isNewGroupCoordinatorEnabled) {
+ val successfulOffsetsPartitions = currentErrors.asScala.filter { case (topicPartition, error) =>
+ topicPartition.topic == GROUP_METADATA_TOPIC_NAME && error == Errors.NONE
+ }.keys
+
+ if (successfulOffsetsPartitions.nonEmpty) {
+ // as soon as the end transaction marker has been written for a transactional offset commit,
+ // call to the group coordinator to materialize the offsets into the cache
+ try {
+ groupCoordinator.onTransactionCompleted(producerId, successfulOffsetsPartitions.asJava, result)
+ } catch {
+ case e: Exception =>
+ error(s"Received an exception while trying to update the offsets cache on transaction marker append", e)
+ val updatedErrors = new ConcurrentHashMap[TopicPartition, Errors]()
+ successfulOffsetsPartitions.foreach(updatedErrors.put(_, Errors.UNKNOWN_SERVER_ERROR))
+ updateErrors(producerId, updatedErrors)
+ }
}
}
@@ -2416,14 +2419,56 @@ class KafkaApis(val requestChannel: RequestChannel,
numAppends.decrementAndGet()
skippedMarkers += 1
} else {
- val controlRecords = partitionsWithCompatibleMessageFormat.map { partition =>
- val controlRecordType = marker.transactionResult match {
- case TransactionResult.COMMIT => ControlRecordType.COMMIT
- case TransactionResult.ABORT => ControlRecordType.ABORT
+ val controlRecordType = marker.transactionResult match {
+ case TransactionResult.COMMIT => ControlRecordType.COMMIT
+ case TransactionResult.ABORT => ControlRecordType.ABORT
+ }
+
+ val markerResults = new ConcurrentHashMap[TopicPartition, Errors]()
+ def maybeComplete(): Unit = {
+ if (partitionsWithCompatibleMessageFormat.size == markerResults.size) {
+ maybeSendResponseCallback(producerId, marker.transactionResult, markerResults)
+ }
+ }
+
+ val controlRecords = mutable.Map.empty[TopicPartition, MemoryRecords]
+ partitionsWithCompatibleMessageFormat.foreach { partition =>
+ if (config.isNewGroupCoordinatorEnabled && partition.topic == GROUP_METADATA_TOPIC_NAME) {
+ // When the new group coordinator is used, writing the end marker is fully delegated
+ // to the group coordinator.
+ groupCoordinator.completeTransaction(
+ partition,
+ marker.producerId,
+ marker.producerEpoch,
+ marker.coordinatorEpoch,
+ marker.transactionResult,
+ Duration.ofMillis(config.requestTimeoutMs.toLong)
+ ).whenComplete { (_, exception) =>
+ val error = if (exception == null) {
+ Errors.NONE
+ } else {
+ Errors.forException(exception) match {
+ case Errors.COORDINATOR_NOT_AVAILABLE | Errors.COORDINATOR_LOAD_IN_PROGRESS | Errors.NOT_COORDINATOR =>
+ // The transaction coordinator does not expect those errors so we translate them
+ // to NOT_LEADER_OR_FOLLOWER to signal to it that the coordinator is not ready yet.
+ Errors.NOT_LEADER_OR_FOLLOWER
+ case error =>
+ error
+ }
+ }
+ markerResults.put(partition, error)
+ maybeComplete()
+ }
+ } else {
+ // Otherwise, the regular appendRecords path is used for all the non __consumer_offsets
+ // partitions or for all partitions when the new group coordinator is disabled.
+ controlRecords += partition -> MemoryRecords.withEndTransactionMarker(
+ producerId,
+ marker.producerEpoch,
+ new EndTransactionMarker(controlRecordType, marker.coordinatorEpoch)
+ )
}
- val endTxnMarker = new EndTransactionMarker(controlRecordType, marker.coordinatorEpoch)
- partition -> MemoryRecords.withEndTransactionMarker(producerId, marker.producerEpoch, endTxnMarker)
- }.toMap
+ }
replicaManager.appendRecords(
timeout = config.requestTimeoutMs.toLong,
@@ -2432,7 +2477,13 @@ class KafkaApis(val requestChannel: RequestChannel,
origin = AppendOrigin.COORDINATOR,
entriesPerPartition = controlRecords,
requestLocal = requestLocal,
- responseCallback = maybeSendResponseCallback(producerId, marker.transactionResult))
+ responseCallback = errors => {
+ errors.forKeyValue { (tp, partitionResponse) =>
+ markerResults.put(tp, partitionResponse.error)
+ }
+ maybeComplete()
+ }
+ )
}
}
@@ -2446,6 +2497,7 @@ class KafkaApis(val requestChannel: RequestChannel,
if (config.interBrokerProtocolVersion.isLessThan(version))
throw new UnsupportedVersionException(s"inter.broker.protocol.version: ${config.interBrokerProtocolVersion.version} is less than the required version: ${version.version}")
}
+
def handleAddPartitionsToTxnRequest(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
ensureInterBrokerVersion(IBP_0_11_0_IV0)
val addPartitionsToTxnRequest =
diff --git a/core/src/test/scala/integration/kafka/api/TransactionsTest.scala b/core/src/test/scala/integration/kafka/api/TransactionsTest.scala
index 10e1e2311de..1c5064d29ab 100644
--- a/core/src/test/scala/integration/kafka/api/TransactionsTest.scala
+++ b/core/src/test/scala/integration/kafka/api/TransactionsTest.scala
@@ -69,8 +69,13 @@ class TransactionsTest extends IntegrationTestHarness {
props.put(KafkaConfig.AutoLeaderRebalanceEnableProp, false.toString)
props.put(KafkaConfig.GroupInitialRebalanceDelayMsProp, "0")
props.put(KafkaConfig.TransactionsAbortTimedOutTransactionCleanupIntervalMsProp, "200")
- props
+ // The new group coordinator does not support verifying transactions yet.
+ if (isNewGroupCoordinatorEnabled()) {
+ props.put(KafkaConfig.TransactionPartitionVerificationEnableProp, "false")
+ }
+
+ props
}
override protected def modifyConfigs(props: Seq[Properties]): Unit = {
@@ -111,7 +116,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testBasicTransactions(quorum: String): Unit = {
val producer = transactionalProducers.head
val consumer = transactionalConsumers.head
@@ -172,7 +177,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testReadCommittedConsumerShouldNotSeeUndecidedData(quorum: String): Unit = {
val producer1 = transactionalProducers.head
val producer2 = createTransactionalProducer("other")
@@ -240,7 +245,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testDelayedFetchIncludesAbortedTransaction(quorum: String): Unit = {
val producer1 = transactionalProducers.head
val producer2 = createTransactionalProducer("other")
@@ -299,14 +304,14 @@ class TransactionsTest extends IntegrationTestHarness {
@nowarn("cat=deprecation")
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testSendOffsetsWithGroupId(quorum: String): Unit = {
sendOffset((producer, groupId, consumer) =>
producer.sendOffsetsToTransaction(TestUtils.consumerPositions(consumer).asJava, groupId))
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testSendOffsetsWithGroupMetadata(quorum: String): Unit = {
sendOffset((producer, _, consumer) =>
producer.sendOffsetsToTransaction(TestUtils.consumerPositions(consumer).asJava, consumer.groupMetadata()))
@@ -386,7 +391,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testFencingOnCommit(quorum: String): Unit = {
val producer1 = transactionalProducers(0)
val producer2 = transactionalProducers(1)
@@ -416,7 +421,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testFencingOnSendOffsets(quorum: String): Unit = {
val producer1 = transactionalProducers(0)
val producer2 = transactionalProducers(1)
@@ -448,7 +453,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testOffsetMetadataInSendOffsetsToTransaction(quorum: String): Unit = {
val tp = new TopicPartition(topic1, 0)
val groupId = "group"
@@ -474,26 +479,26 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testInitTransactionsTimeout(quorum: String): Unit = {
testTimeout(false, producer => producer.initTransactions())
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testSendOffsetsToTransactionTimeout(quorum: String): Unit = {
testTimeout(true, producer => producer.sendOffsetsToTransaction(
Map(new TopicPartition(topic1, 0) -> new OffsetAndMetadata(0)).asJava, new ConsumerGroupMetadata("test-group")))
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testCommitTransactionTimeout(quorum: String): Unit = {
testTimeout(true, producer => producer.commitTransaction())
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testAbortTransactionTimeout(quorum: String): Unit = {
testTimeout(true, producer => producer.abortTransaction())
}
@@ -514,7 +519,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testFencingOnSend(quorum: String): Unit = {
val producer1 = transactionalProducers(0)
val producer2 = transactionalProducers(1)
@@ -559,7 +564,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testFencingOnAddPartitions(quorum: String): Unit = {
val producer1 = transactionalProducers(0)
val producer2 = transactionalProducers(1)
@@ -606,7 +611,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testFencingOnTransactionExpiration(quorum: String): Unit = {
val producer = createTransactionalProducer("expiringProducer", transactionTimeoutMs = 100)
@@ -649,7 +654,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testMultipleMarkersOneLeader(quorum: String): Unit = {
val firstProducer = transactionalProducers.head
val consumer = transactionalConsumers.head
@@ -687,7 +692,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testConsecutivelyRunInitTransactions(quorum: String): Unit = {
val producer = createTransactionalProducer(transactionalId = "normalProducer")
@@ -696,7 +701,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testBumpTransactionalEpoch(quorum: String): Unit = {
val producer = createTransactionalProducer("transactionalProducer",
deliveryTimeoutMs = 5000, requestTimeoutMs = 5000)
@@ -758,7 +763,7 @@ class TransactionsTest extends IntegrationTestHarness {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
+ @ValueSource(strings = Array("zk", "kraft", "kraft+kip848"))
def testFailureToFenceEpoch(quorum: String): Unit = {
val producer1 = transactionalProducers.head
val producer2 = createTransactionalProducer("transactional-producer", maxBlockMs = 1000)
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala
index 9fc10299735..c583455817a 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorLoaderImplTest.scala
@@ -21,7 +21,8 @@ import kafka.server.ReplicaManager
import kafka.utils.TestUtils
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.errors.NotLeaderOrFollowerException
-import org.apache.kafka.common.record.{CompressionType, FileRecords, MemoryRecords, RecordBatch, SimpleRecord}
+import org.apache.kafka.common.record.{CompressionType, ControlRecordType, EndTransactionMarker, FileRecords, MemoryRecords, RecordBatch, SimpleRecord}
+import org.apache.kafka.common.requests.TransactionResult
import org.apache.kafka.common.utils.{MockTime, Time}
import org.apache.kafka.coordinator.group.runtime.CoordinatorLoader.UnknownRecordTypeException
import org.apache.kafka.coordinator.group.runtime.{CoordinatorLoader, CoordinatorPlayback}
@@ -105,7 +106,7 @@ class CoordinatorLoaderImplTest {
)) { loader =>
when(replicaManager.getLog(tp)).thenReturn(Some(log))
when(log.logStartOffset).thenReturn(0L)
- when(replicaManager.getLogEndOffset(tp)).thenReturn(Some(7L))
+ when(replicaManager.getLogEndOffset(tp)).thenReturn(Some(9L))
when(log.highWatermark).thenReturn(0L)
val readResult1 = logReadResult(startOffset = 0, records = Seq(
@@ -145,6 +146,34 @@ class CoordinatorLoaderImplTest {
minOneMessage = true
)).thenReturn(readResult3)
+ val readResult4 = logReadResult(
+ startOffset = 7,
+ producerId = 100L,
+ producerEpoch = 5,
+ controlRecordType = ControlRecordType.COMMIT
+ )
+
+ when(log.read(
+ startOffset = 7L,
+ maxLength = 1000,
+ isolation = FetchIsolation.LOG_END,
+ minOneMessage = true
+ )).thenReturn(readResult4)
+
+ val readResult5 = logReadResult(
+ startOffset = 8,
+ producerId = 500L,
+ producerEpoch = 10,
+ controlRecordType = ControlRecordType.ABORT
+ )
+
+ when(log.read(
+ startOffset = 8L,
+ maxLength = 1000,
+ isolation = FetchIsolation.LOG_END,
+ minOneMessage = true
+ )).thenReturn(readResult5)
+
assertNotNull(loader.load(tp, coordinator).get(10, TimeUnit.SECONDS))
verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k1", "v1"))
@@ -154,8 +183,12 @@ class CoordinatorLoaderImplTest {
verify(coordinator).replay(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, ("k5", "v5"))
verify(coordinator).replay(100L, 5.toShort, ("k6", "v6"))
verify(coordinator).replay(100L, 5.toShort, ("k7", "v7"))
+ verify(coordinator).replayEndTransactionMarker(100L, 5, TransactionResult.COMMIT)
+ verify(coordinator).replayEndTransactionMarker(500L, 10, TransactionResult.ABORT)
verify(coordinator).updateLastWrittenOffset(2)
verify(coordinator).updateLastWrittenOffset(5)
+ verify(coordinator).updateLastWrittenOffset(7)
+ verify(coordinator).updateLastWrittenOffset(8)
verify(coordinator).updateLastCommittedOffset(0)
}
}
@@ -635,4 +668,35 @@ class CoordinatorLoaderImplTest {
new FetchDataInfo(new LogOffsetMetadata(startOffset), fileRecords)
}
+
+ private def logReadResult(
+ startOffset: Long,
+ producerId: Long,
+ producerEpoch: Short,
+ controlRecordType: ControlRecordType
+ ): FetchDataInfo = {
+ val fileRecords = mock(classOf[FileRecords])
+ val memoryRecords = MemoryRecords.withEndTransactionMarker(
+ startOffset,
+ 0L,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH,
+ producerId,
+ producerEpoch,
+ new EndTransactionMarker(controlRecordType, 0)
+ )
+
+ when(fileRecords.sizeInBytes).thenReturn(memoryRecords.sizeInBytes)
+
+ val bufferCapture: ArgumentCaptor[ByteBuffer] = ArgumentCaptor.forClass(classOf[ByteBuffer])
+ when(fileRecords.readInto(
+ bufferCapture.capture(),
+ ArgumentMatchers.anyInt())
+ ).thenAnswer { _ =>
+ val buffer = bufferCapture.getValue
+ buffer.put(memoryRecords.buffer.duplicate)
+ buffer.flip()
+ }
+
+ new FetchDataInfo(new LogOffsetMetadata(startOffset), fileRecords)
+ }
}
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala
index 121a1f119a1..0cfbb8f821c 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala
@@ -22,13 +22,16 @@ import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.errors.{NotLeaderOrFollowerException, RecordTooLargeException}
import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch}
+import org.apache.kafka.common.record.{CompressionType, ControlRecordType, MemoryRecords, RecordBatch}
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.requests.TransactionResult
import org.apache.kafka.common.utils.{MockTime, Time}
import org.apache.kafka.coordinator.group.runtime.PartitionWriter
import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig}
import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue}
import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.EnumSource
import org.mockito.{ArgumentCaptor, ArgumentMatchers}
import org.mockito.Mockito.{mock, verify, when}
@@ -234,6 +237,80 @@ class CoordinatorPartitionWriterTest {
assertEquals(records, receivedRecords)
}
+ @ParameterizedTest
+ @EnumSource(value = classOf[ControlRecordType], names = Array("COMMIT", "ABORT"))
+ def testWriteEndTransactionMarker(controlRecordType: ControlRecordType): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val time = new MockTime()
+ val partitionRecordWriter = new CoordinatorPartitionWriter(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ time
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ val recordsCapture: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ val callbackCapture: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit])
+
+ when(replicaManager.appendRecords(
+ ArgumentMatchers.eq(0L),
+ ArgumentMatchers.eq(1.toShort),
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ recordsCapture.capture(),
+ callbackCapture.capture(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenAnswer(_ => {
+ callbackCapture.getValue.apply(Map(
+ tp -> new PartitionResponse(
+ Errors.NONE,
+ 5,
+ 10,
+ RecordBatch.NO_TIMESTAMP,
+ -1,
+ Collections.emptyList(),
+ ""
+ )
+ ))
+ })
+
+ assertEquals(11, partitionRecordWriter.appendEndTransactionMarker(
+ tp,
+ 100L,
+ 50.toShort,
+ 10,
+ if (controlRecordType == ControlRecordType.COMMIT) TransactionResult.COMMIT else TransactionResult.ABORT
+ ))
+
+ val batch = recordsCapture.getValue.getOrElse(tp,
+ throw new AssertionError(s"No records for $tp"))
+ assertEquals(1, batch.batches.asScala.toList.size)
+
+ val firstBatch = batch.batches.asScala.head
+ assertEquals(100L, firstBatch.producerId)
+ assertEquals(50.toShort, firstBatch.producerEpoch)
+ assertTrue(firstBatch.isTransactional)
+ assertTrue(firstBatch.isControlBatch)
+
+ val receivedRecords = batch.records.asScala.map { record =>
+ ControlRecordType.parse(record.key)
+ }.toList
+
+ assertEquals(List(controlRecordType), receivedRecords)
+ }
+
@Test
def testWriteRecordsWithFailure(): Unit = {
val tp = new TopicPartition("foo", 0)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index 702fd1de88a..26e3670b242 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -22,7 +22,7 @@ import java.nio.charset.StandardCharsets
import java.util
import java.util.Arrays.asList
import java.util.concurrent.{CompletableFuture, TimeUnit}
-import java.util.{Collections, Optional, OptionalInt, OptionalLong, Properties}
+import java.util.{Collections, Comparator, Optional, OptionalInt, OptionalLong, Properties}
import kafka.api.LeaderAndIsr
import kafka.cluster.{Broker, Partition}
import kafka.controller.{ControllerContext, KafkaController}
@@ -84,7 +84,7 @@ import org.apache.kafka.server.authorizer.{Action, AuthorizationResult, Authoriz
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test}
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.{CsvSource, ValueSource}
+import org.junit.jupiter.params.provider.{CsvSource, EnumSource, ValueSource}
import org.mockito.ArgumentMatchers.{any, anyBoolean, anyDouble, anyInt, anyLong, anyShort, anyString, argThat, isNotNull}
import org.mockito.Mockito.{mock, reset, times, verify, when}
import org.mockito.{ArgumentCaptor, ArgumentMatchers, Mockito}
@@ -100,9 +100,11 @@ import org.apache.kafka.server.ClientMetricsManager
import org.apache.kafka.server.common.{Features, MetadataVersion}
import org.apache.kafka.server.common.MetadataVersion.{IBP_0_10_2_IV0, IBP_2_2_IV1}
import org.apache.kafka.server.metrics.ClientMetricsTestUtils
-import org.apache.kafka.server.util.MockTime
+import org.apache.kafka.server.util.{FutureUtils, MockTime}
import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchParams, FetchPartitionData, LogConfig}
+import java.time.Duration
+
class KafkaApisTest {
private val requestChannel: RequestChannel = mock(classOf[RequestChannel])
private val requestChannelMetrics: RequestChannel.Metrics = mock(classOf[RequestChannel.Metrics])
@@ -3018,6 +3020,223 @@ class KafkaApisTest {
any())
}
+ @Test
+ def testHandleWriteTxnMarkersRequestWithNewGroupCoordinator(): Unit = {
+ val offset0 = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0)
+ val offset1 = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 1)
+ val foo0 = new TopicPartition("foo", 0)
+ val foo1 = new TopicPartition("foo", 1)
+
+ val allPartitions = List(
+ offset0,
+ offset1,
+ foo0,
+ foo1
+ )
+
+ val writeTxnMarkersRequest = new WriteTxnMarkersRequest.Builder(
+ ApiKeys.WRITE_TXN_MARKERS.latestVersion(),
+ List(
+ new TxnMarkerEntry(
+ 1L,
+ 1.toShort,
+ 0,
+ TransactionResult.COMMIT,
+ List(offset0, foo0).asJava
+ ),
+ new TxnMarkerEntry(
+ 2L,
+ 1.toShort,
+ 0,
+ TransactionResult.ABORT,
+ List(offset1, foo1).asJava
+ )
+ ).asJava
+ ).build()
+
+ val requestChannelRequest = buildRequest(writeTxnMarkersRequest)
+
+ allPartitions.foreach { tp =>
+ when(replicaManager.getMagic(tp))
+ .thenReturn(Some(RecordBatch.MAGIC_VALUE_V2))
+ }
+
+ when(groupCoordinator.completeTransaction(
+ ArgumentMatchers.eq(offset0),
+ ArgumentMatchers.eq(1L),
+ ArgumentMatchers.eq(1.toShort),
+ ArgumentMatchers.eq(0),
+ ArgumentMatchers.eq(TransactionResult.COMMIT),
+ ArgumentMatchers.eq(Duration.ofMillis(Defaults.RequestTimeoutMs))
+ )).thenReturn(CompletableFuture.completedFuture[Void](null))
+
+ when(groupCoordinator.completeTransaction(
+ ArgumentMatchers.eq(offset1),
+ ArgumentMatchers.eq(2L),
+ ArgumentMatchers.eq(1.toShort),
+ ArgumentMatchers.eq(0),
+ ArgumentMatchers.eq(TransactionResult.ABORT),
+ ArgumentMatchers.eq(Duration.ofMillis(Defaults.RequestTimeoutMs))
+ )).thenReturn(CompletableFuture.completedFuture[Void](null))
+
+ val entriesPerPartition: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ val responseCallback: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit])
+
+ when(replicaManager.appendRecords(
+ ArgumentMatchers.eq(Defaults.RequestTimeoutMs.toLong),
+ ArgumentMatchers.eq(-1),
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ entriesPerPartition.capture(),
+ responseCallback.capture(),
+ any(),
+ any(),
+ ArgumentMatchers.eq(RequestLocal.NoCaching),
+ any(),
+ any()
+ )).thenAnswer { _ =>
+ responseCallback.getValue.apply(
+ entriesPerPartition.getValue.keySet.map { tp =>
+ tp -> new PartitionResponse(Errors.NONE)
+ }.toMap
+ )
+ }
+
+ createKafkaApis(overrideProperties = Map(
+ KafkaConfig.NewGroupCoordinatorEnableProp -> "true"
+ )).handleWriteTxnMarkersRequest(requestChannelRequest, RequestLocal.NoCaching)
+
+ val expectedResponse = new WriteTxnMarkersResponseData()
+ .setMarkers(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerResult()
+ .setProducerId(1L)
+ .setTopics(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult()
+ .setName(Topic.GROUP_METADATA_TOPIC_NAME)
+ .setPartitions(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult()
+ .setPartitionIndex(0)
+ .setErrorCode(Errors.NONE.code)
+ ).asJava),
+ new WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult()
+ .setName("foo")
+ .setPartitions(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult()
+ .setPartitionIndex(0)
+ .setErrorCode(Errors.NONE.code)
+ ).asJava)
+ ).asJava),
+ new WriteTxnMarkersResponseData.WritableTxnMarkerResult()
+ .setProducerId(2L)
+ .setTopics(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult()
+ .setName(Topic.GROUP_METADATA_TOPIC_NAME)
+ .setPartitions(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult()
+ .setPartitionIndex(1)
+ .setErrorCode(Errors.NONE.code)
+ ).asJava),
+ new WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult()
+ .setName("foo")
+ .setPartitions(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult()
+ .setPartitionIndex(1)
+ .setErrorCode(Errors.NONE.code)
+ ).asJava)
+ ).asJava)
+ ).asJava)
+
+ val response = verifyNoThrottling[WriteTxnMarkersResponse](requestChannelRequest)
+ assertEquals(normalize(expectedResponse), normalize(response.data))
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = classOf[Errors], names = Array(
+ "COORDINATOR_NOT_AVAILABLE",
+ "COORDINATOR_LOAD_IN_PROGRESS",
+ "NOT_COORDINATOR",
+ "REQUEST_TIMED_OUT"
+ ))
+ def testHandleWriteTxnMarkersRequestWithNewGroupCoordinatorErrorTranslation(error: Errors): Unit = {
+ val offset0 = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0)
+
+ val writeTxnMarkersRequest = new WriteTxnMarkersRequest.Builder(
+ ApiKeys.WRITE_TXN_MARKERS.latestVersion(),
+ List(
+ new TxnMarkerEntry(
+ 1L,
+ 1.toShort,
+ 0,
+ TransactionResult.COMMIT,
+ List(offset0).asJava
+ )
+ ).asJava
+ ).build()
+
+ val requestChannelRequest = buildRequest(writeTxnMarkersRequest)
+
+ when(replicaManager.getMagic(offset0))
+ .thenReturn(Some(RecordBatch.MAGIC_VALUE_V2))
+
+ when(groupCoordinator.completeTransaction(
+ ArgumentMatchers.eq(offset0),
+ ArgumentMatchers.eq(1L),
+ ArgumentMatchers.eq(1.toShort),
+ ArgumentMatchers.eq(0),
+ ArgumentMatchers.eq(TransactionResult.COMMIT),
+ ArgumentMatchers.eq(Duration.ofMillis(Defaults.RequestTimeoutMs))
+ )).thenReturn(FutureUtils.failedFuture[Void](error.exception()))
+
+ createKafkaApis(overrideProperties = Map(
+ KafkaConfig.NewGroupCoordinatorEnableProp -> "true"
+ )).handleWriteTxnMarkersRequest(requestChannelRequest, RequestLocal.NoCaching)
+
+ val expectedError = error match {
+ case Errors.COORDINATOR_NOT_AVAILABLE | Errors.COORDINATOR_LOAD_IN_PROGRESS | Errors.NOT_COORDINATOR =>
+ Errors.NOT_LEADER_OR_FOLLOWER
+ case error =>
+ error
+ }
+
+ val expectedResponse = new WriteTxnMarkersResponseData()
+ .setMarkers(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerResult()
+ .setProducerId(1L)
+ .setTopics(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult()
+ .setName(Topic.GROUP_METADATA_TOPIC_NAME)
+ .setPartitions(List(
+ new WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult()
+ .setPartitionIndex(0)
+ .setErrorCode(expectedError.code)
+ ).asJava)
+ ).asJava)
+ ).asJava)
+
+ val response = verifyNoThrottling[WriteTxnMarkersResponse](requestChannelRequest)
+ assertEquals(normalize(expectedResponse), normalize(response.data))
+ }
+
+ private def normalize(
+ response: WriteTxnMarkersResponseData
+ ): WriteTxnMarkersResponseData = {
+ val copy = response.duplicate()
+ copy.markers.sort(
+ Comparator.comparingLong[WriteTxnMarkersResponseData.WritableTxnMarkerResult](_.producerId)
+ )
+ copy.markers.forEach { marker =>
+ marker.topics.sort((t1, t2) => t1.name.compareTo(t2.name))
+ marker.topics.forEach { topic =>
+ topic.partitions.sort(
+ Comparator.comparingInt[WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult](_.partitionIndex)
+ )
+ }
+ }
+ copy
+ }
+
@Test
def testLeaderReplicaIfLocalRaisesFencedLeaderEpoch(): Unit = {
testListOffsetFailedGetLeaderReplica(Errors.FENCED_LEADER_EPOCH)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
index 948a58516c0..33fef01ec6b 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
@@ -46,6 +46,7 @@ import org.apache.kafka.common.utils.BufferSupplier;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
+import java.time.Duration;
import java.util.List;
import java.util.OptionalInt;
import java.util.Properties;
@@ -266,6 +267,29 @@ public interface GroupCoordinator {
BufferSupplier bufferSupplier
);
+ /**
+ * Complete a transaction. This is called when the WriteTxnMarkers API is called
+ * by the Transaction Coordinator in order to write the markers to the
+ * __consumer_offsets partitions.
+ *
+ * @param tp The topic-partition.
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param coordinatorEpoch The epoch of the transaction coordinator.
+ * @param result The transaction result.
+ * @param timeout The operation timeout.
+ *
+ * @return A future yielding the result.
+ */
+ CompletableFuture<Void> completeTransaction(
+ TopicPartition tp,
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result,
+ Duration timeout
+ );
+
/**
* Return the partition index for the given Group.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index feb4a92a72a..3e007b74249 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -78,6 +78,7 @@ import org.apache.kafka.coordinator.group.runtime.PartitionWriter;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.record.BrokerCompressionType;
+import org.apache.kafka.server.util.FutureUtils;
import org.apache.kafka.server.util.timer.Timer;
import org.slf4j.Logger;
@@ -902,9 +903,12 @@ public class GroupCoordinatorService implements GroupCoordinator {
));
}
- return runtime.scheduleWriteOperation(
+ return runtime.scheduleTransactionalWriteOperation(
"txn-commit-offset",
topicPartitionFor(request.groupId()),
+ request.transactionalId(),
+ request.producerId(),
+ request.producerEpoch(),
Duration.ofMillis(config.offsetCommitTimeoutMs),
coordinator -> coordinator.commitTransactionalOffset(context, request)
).exceptionally(exception ->
@@ -944,6 +948,39 @@ public class GroupCoordinatorService implements GroupCoordinator {
);
}
+ /**
+ * See {@link GroupCoordinator#completeTransaction(TopicPartition, long, short, int, TransactionResult, Duration)}.
+ */
+ @Override
+ public CompletableFuture<Void> completeTransaction(
+ TopicPartition tp,
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result,
+ Duration timeout
+ ) {
+ if (!isActive.get()) {
+ return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
+ }
+
+ if (!tp.topic().equals(Topic.GROUP_METADATA_TOPIC_NAME)) {
+ return FutureUtils.failedFuture(new IllegalStateException(
+ "Completing a transaction for " + tp + " is not expected"
+ ));
+ }
+
+ return runtime.scheduleTransactionCompletion(
+ "write-txn-marker",
+ tp,
+ producerId,
+ producerEpoch,
+ coordinatorEpoch,
+ result,
+ timeout
+ );
+ }
+
/**
* See {@link GroupCoordinator#onTransactionCompleted(long, Iterable, TransactionResult)}.
*/
@@ -954,6 +991,7 @@ public class GroupCoordinatorService implements GroupCoordinator {
TransactionResult transactionResult
) {
throwIfNotActive();
+ throw new IllegalStateException("onTransactionCompleted is not supported.");
}
/**
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
index b68549ef157..66694afa0c0 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
@@ -43,6 +43,7 @@ import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
import org.apache.kafka.common.protocol.ApiMessage;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.RequestContext;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentKey;
@@ -716,4 +717,21 @@ public class GroupCoordinatorShard implements CoordinatorShard<Record> {
+ " in " + record);
}
}
+
+ /**
+ * Applies the given transaction marker.
+ *
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param result The result of the transaction.
+ * @throws RuntimeException if the transaction can not be completed.
+ */
+ @Override
+ public void replayEndTransactionMarker(
+ long producerId,
+ short producerEpoch,
+ TransactionResult result
+ ) throws RuntimeException {
+ offsetMetadataManager.replayEndTransactionMarker(producerId, result);
+ }
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
index 0de78b3ef92..b4f90bd1bd0 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
@@ -36,6 +36,7 @@ import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.OffsetCommitRequest;
import org.apache.kafka.common.requests.RequestContext;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.coordinator.group.generated.OffsetCommitKey;
@@ -861,6 +862,7 @@ public class OffsetMetadataManager {
}
if (producerId == RecordBatch.NO_PRODUCER_ID) {
+ log.debug("Replaying offset commit with key {}, value {}", key, value);
// If the offset is not part of a transaction, it is directly stored
// in the offsets store.
OffsetAndMetadata previousValue = offsets.put(
@@ -873,6 +875,7 @@ public class OffsetMetadataManager {
metrics.incrementNumOffsets();
}
} else {
+ log.debug("Replaying transactional offset commit with producer id {}, key {}, value {}", producerId, key, value);
// Otherwise, the transaction offset is stored in the pending transactional
// offsets store. Pending offsets there are moved to the main store when
// the transaction is committed; or removed when the transaction is aborted.
@@ -891,6 +894,43 @@ public class OffsetMetadataManager {
}
}
+ /**
+ * Applies the given transaction marker.
+ *
+ * @param producerId The producer id.
+ * @param result The result of the transaction.
+ * @throws RuntimeException if the transaction can not be completed.
+ */
+ public void replayEndTransactionMarker(
+ long producerId,
+ TransactionResult result
+ ) throws RuntimeException {
+ Offsets pendingOffsets = pendingTransactionalOffsets.remove(producerId);
+
+ if (result == TransactionResult.COMMIT) {
+ log.debug("Committed transactional offset commits for producer id {}.", producerId);
+ if (pendingOffsets == null) return;
+
+ pendingOffsets.offsetsByGroup.forEach((groupId, topicOffsets) -> {
+ topicOffsets.forEach((topicName, partitionOffsets) -> {
+ partitionOffsets.forEach((partitionId, offsetAndMetadata) -> {
+ log.debug("Committed transaction offset commit for producer id {} in group {} " +
+ "with topic {}, partition {}, and offset {}.",
+ producerId, groupId, topicName, partitionId, offsetAndMetadata);
+ offsets.put(
+ groupId,
+ topicName,
+ partitionId,
+ offsetAndMetadata
+ );
+ });
+ });
+ });
+ } else {
+ log.debug("Aborted transactional offset commits for producer id {}.", producerId);
+ }
+ }
+
/**
* A new metadata image is available.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorPlayback.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorPlayback.java
index 1fd8c9712a6..9aa5400cc05 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorPlayback.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorPlayback.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.coordinator.group.runtime;
+import org.apache.kafka.common.requests.TransactionResult;
+
/**
* The CoordinatorPlayback interface. This interface is used to replay
* records to the coordinator in order to update its state. This is
@@ -39,6 +41,20 @@ public interface CoordinatorPlayback<U> {
U record
) throws RuntimeException;
+ /**
+ * Applies the given transaction marker.
+ *
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param result The result of the transaction.
+ * @throws RuntimeException if the transaction can not be completed.
+ */
+ void replayEndTransactionMarker(
+ long producerId,
+ short producerEpoch,
+ TransactionResult result
+ ) throws RuntimeException;
+
/**
* Invoke operations when a batch has been successfully loaded.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
index 75e9f1d478b..fffdd83b256 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
@@ -23,6 +23,7 @@ import org.apache.kafka.common.errors.NotCoordinatorException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
@@ -913,6 +914,164 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
}
}
+ /**
+ * A coordinator event that applies and writes a transaction end marker.
+ */
+ class CoordinatorCompleteTransactionEvent implements CoordinatorEvent, DeferredEvent {
+ /**
+ * The topic partition that this write event is applied to.
+ */
+ final TopicPartition tp;
+
+ /**
+ * The operation name.
+ */
+ final String name;
+
+ /**
+ * The producer id.
+ */
+ final long producerId;
+
+ /**
+ * The producer epoch.
+ */
+ final short producerEpoch;
+
+ /**
+ * The coordinator epoch of the transaction coordinator.
+ */
+ final int coordinatorEpoch;
+
+ /**
+ * The transaction result.
+ */
+ final TransactionResult result;
+
+ /**
+ * Timeout value for the write operation.
+ */
+ final Duration writeTimeout;
+
+ /**
+ * The future that will be completed with the response
+ * generated by the write operation or an error.
+ */
+ final CompletableFuture<Void> future;
+
+ /**
+ * The time this event was created.
+ */
+ private final long createdTimeMs;
+
+ CoordinatorCompleteTransactionEvent(
+ String name,
+ TopicPartition tp,
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result,
+ Duration writeTimeout
+ ) {
+ this.name = name;
+ this.tp = tp;
+ this.producerId = producerId;
+ this.producerEpoch = producerEpoch;
+ this.coordinatorEpoch = coordinatorEpoch;
+ this.result = result;
+ this.writeTimeout = writeTimeout;
+ this.future = new CompletableFuture<>();
+ this.createdTimeMs = time.milliseconds();
+ }
+
+ /**
+ * @return The key used by the CoordinatorEventProcessor to ensure
+ * that events with the same key are not processed concurrently.
+ */
+ @Override
+ public TopicPartition key() {
+ return tp;
+ }
+
+ /**
+ * Called by the CoordinatorEventProcessor when the event is executed.
+ */
+ @Override
+ public void run() {
+ try {
+ withActiveContextOrThrow(tp, context -> {
+ long prevLastWrittenOffset = context.coordinator.lastWrittenOffset();
+
+ try {
+ context.coordinator.replayEndTransactionMarker(
+ producerId,
+ producerEpoch,
+ result
+ );
+
+ long offset = partitionWriter.appendEndTransactionMarker(
+ tp,
+ producerId,
+ producerEpoch,
+ coordinatorEpoch,
+ result
+ );
+ context.coordinator.updateLastWrittenOffset(offset);
+
+ if (!future.isDone()) {
+ context.deferredEventQueue.add(offset, this);
+ timer.add(new TimerTask(writeTimeout.toMillis()) {
+ @Override
+ public void run() {
+ if (!future.isDone()) {
+ scheduleInternalOperation(
+ "WriteTimeout(name=" + name + ", tp=" + tp + ")",
+ tp,
+ () -> complete(new TimeoutException("CoordinatorCompleteTransactionEvent " + name +
+ " timed out after " + writeTimeout.toMillis() + "ms"))
+ );
+ }
+ }
+ });
+ } else {
+ complete(null);
+ }
+ } catch (Throwable t) {
+ context.coordinator.revertLastWrittenOffset(prevLastWrittenOffset);
+ complete(t);
+ }
+ });
+ } catch (Throwable t) {
+ complete(t);
+ }
+ }
+
+ /**
+ * Completes the future with either the result of the write operation
+ * or the provided exception.
+ *
+ * @param exception The exception to complete the future with.
+ */
+ @Override
+ public void complete(Throwable exception) {
+ if (exception == null) {
+ future.complete(null);
+ } else {
+ future.completeExceptionally(exception);
+ }
+ }
+
+ @Override
+ public long createdTimeMs() {
+ return createdTimeMs;
+ }
+
+ @Override
+ public String toString() {
+ return "CoordinatorCompleteTransactionEvent(name=" + name + ")";
+ }
+ }
+
/**
* A coordinator internal event.
*/
@@ -1320,6 +1479,44 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
return event.future;
}
+ /**
+ * Schedules the transaction completion.
+ *
+ * @param name The name of the operation.
+ * @param tp The address of the coordinator (aka its topic-partitions).
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param coordinatorEpoch The epoch of the transaction coordinator.
+ * @param result The transaction result.
+ *
+ * @return A future that will be completed with null when the operation is
+ * completed or an exception if the operation failed.
+ */
+ public CompletableFuture<Void> scheduleTransactionCompletion(
+ String name,
+ TopicPartition tp,
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result,
+ Duration timeout
+ ) {
+ throwIfNotRunning();
+ log.debug("Scheduled execution of transaction completion for {} with producer id={}, producer epoch={}, " +
+ "coordinator epoch={} and transaction result={}.", tp, producerId, producerEpoch, coordinatorEpoch, result);
+ CoordinatorCompleteTransactionEvent event = new CoordinatorCompleteTransactionEvent(
+ name,
+ tp,
+ producerId,
+ producerEpoch,
+ coordinatorEpoch,
+ result,
+ timeout
+ );
+ enqueue(event);
+ return event.future;
+ }
+
/**
* Schedules a read operation.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorShard.java
index cd5f4e258b4..391acff144f 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorShard.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorShard.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.coordinator.group.runtime;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
@@ -51,11 +52,27 @@ public interface CoordinatorShard<U> {
/**
* Replay a record to update the state machine.
*
- * @param record The record to replay.
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param record The record to replay.
*/
void replay(
long producerId,
short producerEpoch,
U record
) throws RuntimeException;
+
+ /**
+ * Applies the end transaction marker.
+ *
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param result The result of the transaction.
+ * @throws RuntimeException if the transaction can not be completed.
+ */
+ default void replayEndTransactionMarker(
+ long producerId,
+ short producerEpoch,
+ TransactionResult result
+ ) throws RuntimeException {}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java
index e3efbfaf7b2..e4270dd8fcb 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java
@@ -18,6 +18,7 @@ package org.apache.kafka.coordinator.group.runtime;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.requests.TransactionResult;
import java.util.List;
@@ -96,4 +97,23 @@ public interface PartitionWriter<T> {
short producerEpoch,
List<T> records
) throws KafkaException;
+
+ /**
+ * Write the transaction end marker.
+ *
+ * @param tp The partition to write records to.
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param coordinatorEpoch The epoch of the transaction coordinator.
+ * @param result The transaction result.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ long appendEndTransactionMarker(
+ TopicPartition tp,
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result
+ ) throws KafkaException;
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java
index 63513465569..c5c6995c37d 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/SnapshottableCoordinator.java
@@ -17,6 +17,7 @@
package org.apache.kafka.coordinator.group.runtime;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
@@ -111,6 +112,22 @@ class SnapshottableCoordinator<S extends CoordinatorShard<U>, U> implements Coor
coordinator.replay(producerId, producerEpoch, record);
}
+ /**
+ * Applies the end transaction marker.
+ *
+ * @param producerId The producer id.
+ * @param producerEpoch The producer epoch.
+ * @param result The result of the transaction.
+ */
+ @Override
+ public synchronized void replayEndTransactionMarker(
+ long producerId,
+ short producerEpoch,
+ TransactionResult result
+ ) {
+ coordinator.replayEndTransactionMarker(producerId, producerEpoch, result);
+ }
+
/**
* Updates the last written offset. This also create a new snapshot
* in the snapshot registry.
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
index 719d1660670..b9b162dc647 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
@@ -59,6 +59,7 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.common.requests.RequestHeader;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.utils.BufferSupplier;
@@ -93,8 +94,10 @@ import java.util.stream.Stream;
import static org.apache.kafka.common.requests.JoinGroupRequest.UNKNOWN_MEMBER_ID;
import static org.apache.kafka.coordinator.group.TestUtil.requestContext;
+import static org.apache.kafka.test.TestUtils.assertFutureThrows;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -1870,6 +1873,8 @@ public class GroupCoordinatorServiceTest {
TxnOffsetCommitRequestData request = new TxnOffsetCommitRequestData()
.setGroupId("foo")
.setTransactionalId("transactional-id")
+ .setProducerId(10L)
+ .setProducerEpoch((short) 5)
.setMemberId("member-id")
.setGenerationId(10)
.setTopics(Collections.singletonList(new TxnOffsetCommitRequestData.TxnOffsetCommitRequestTopic()
@@ -1885,9 +1890,12 @@ public class GroupCoordinatorServiceTest {
.setPartitionIndex(0)
.setErrorCode(Errors.NONE.code())))));
- when(runtime.scheduleWriteOperation(
+ when(runtime.scheduleTransactionalWriteOperation(
ArgumentMatchers.eq("txn-commit-offset"),
ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.eq("transactional-id"),
+ ArgumentMatchers.eq(10L),
+ ArgumentMatchers.eq((short) 5),
ArgumentMatchers.eq(Duration.ofMillis(5000)),
ArgumentMatchers.any()
)).thenReturn(CompletableFuture.completedFuture(response));
@@ -1900,4 +1908,82 @@ public class GroupCoordinatorServiceTest {
assertEquals(response, future.get());
}
+
+ @Test
+ public void testCompleteTransaction() throws ExecutionException, InterruptedException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime,
+ new GroupCoordinatorMetrics()
+ );
+ service.startup(() -> 1);
+
+ when(runtime.scheduleTransactionCompletion(
+ ArgumentMatchers.eq("write-txn-marker"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.eq(100L),
+ ArgumentMatchers.eq((short) 5),
+ ArgumentMatchers.eq(10),
+ ArgumentMatchers.eq(TransactionResult.COMMIT),
+ ArgumentMatchers.eq(Duration.ofMillis(100))
+ )).thenReturn(CompletableFuture.completedFuture(null));
+
+ CompletableFuture<Void> future = service.completeTransaction(
+ new TopicPartition("__consumer_offsets", 0),
+ 100L,
+ (short) 5,
+ 10,
+ TransactionResult.COMMIT,
+ Duration.ofMillis(100)
+ );
+
+ assertNull(future.get());
+ }
+
+ @Test
+ public void testCompleteTransactionWhenNotCoordinatorServiceStarted() {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime,
+ new GroupCoordinatorMetrics()
+ );
+
+ CompletableFuture<Void> future = service.completeTransaction(
+ new TopicPartition("foo", 0),
+ 100L,
+ (short) 5,
+ 10,
+ TransactionResult.COMMIT,
+ Duration.ofMillis(100)
+ );
+
+ assertFutureThrows(future, CoordinatorNotAvailableException.class);
+ }
+
+ @Test
+ public void testCompleteTransactionWithUnexpectedPartition() {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime,
+ new GroupCoordinatorMetrics()
+ );
+ service.startup(() -> 1);
+
+ CompletableFuture<Void> future = service.completeTransaction(
+ new TopicPartition("foo", 0),
+ 100L,
+ (short) 5,
+ 10,
+ TransactionResult.COMMIT,
+ Duration.ofMillis(100)
+ );
+
+ assertFutureThrows(future, IllegalStateException.class);
+ }
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
index 9d0bccc012b..740ad2413c1 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
@@ -27,6 +27,7 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.RequestContext;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
@@ -52,6 +53,8 @@ import org.apache.kafka.coordinator.group.runtime.CoordinatorResult;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatchers;
@@ -968,4 +971,34 @@ public class GroupCoordinatorShardTest {
verify(groupMetadataManager, times(1)).maybeDeleteGroup(eq("group-id"), any());
verify(groupMetadataManager, times(0)).maybeDeleteGroup(eq("other-group-id"), any());
}
+
+ @ParameterizedTest
+ @EnumSource(value = TransactionResult.class)
+ public void testReplayEndTransactionMarker(TransactionResult result) {
+ GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
+ CoordinatorMetrics coordinatorMetrics = mock(CoordinatorMetrics.class);
+ CoordinatorMetricsShard metricsShard = mock(CoordinatorMetricsShard.class);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ Time.SYSTEM,
+ new MockCoordinatorTimer<>(Time.SYSTEM),
+ mock(GroupCoordinatorConfig.class),
+ coordinatorMetrics,
+ metricsShard
+ );
+
+ coordinator.replayEndTransactionMarker(
+ 100L,
+ (short) 5,
+ result
+ );
+
+ verify(offsetMetadataManager, times(1)).replayEndTransactionMarker(
+ 100L,
+ result
+ );
+ }
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
index bf1ef890fff..b86ddc728b1 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
@@ -41,6 +41,7 @@ import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.common.requests.RequestHeader;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.utils.LogContext;
@@ -442,6 +443,15 @@ public class OffsetMetadataManagerTest {
lastWrittenOffset++;
}
+ private void replayEndTransactionMarker(
+ long producerId,
+ TransactionResult result
+ ) {
+ snapshotRegistry.getOrCreateSnapshot(lastWrittenOffset);
+ offsetMetadataManager.replayEndTransactionMarker(producerId, result);
+ lastWrittenOffset++;
+ }
+
public void testOffsetDeleteWith(
String groupId,
String topic,
@@ -2414,6 +2424,73 @@ public class OffsetMetadataManagerTest {
assertNull(context.offsetMetadataManager.offset("foo", "bar", 0));
}
+ @Test
+ public void testReplayTransactionEndMarkerWithCommit() {
+ OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
+
+ // Add pending transactional commit for producer id 5.
+ verifyTransactionalReplay(context, 5L, "foo", "bar", 0, new OffsetAndMetadata(
+ 100L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Add pending transactional commit for producer id 6.
+ verifyTransactionalReplay(context, 6L, "foo", "bar", 1, new OffsetAndMetadata(
+ 200L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Replaying an end marker with an unknown producer id should not fail.
+ context.replayEndTransactionMarker(1L, TransactionResult.COMMIT);
+
+ // Replaying an end marker to commit transaction of producer id 5.
+ context.replayEndTransactionMarker(5L, TransactionResult.COMMIT);
+
+ // The pending offset is removed...
+ assertNull(context.offsetMetadataManager.pendingTransactionalOffset(
+ 5L,
+ "foo",
+ "bar",
+ 0
+ ));
+
+ // ... and added to the main offset storage.
+ assertEquals(new OffsetAndMetadata(
+ 100L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ), context.offsetMetadataManager.offset(
+ "foo",
+ "bar",
+ 0
+ ));
+
+ // Replaying an end marker to abort transaction of producer id 6.
+ context.replayEndTransactionMarker(6L, TransactionResult.ABORT);
+
+ // The pending offset is removed from the pending offsets and
+ // it is not added to the main offset storage.
+ assertNull(context.offsetMetadataManager.pendingTransactionalOffset(
+ 6L,
+ "foo",
+ "bar",
+ 1
+ ));
+ assertNull(context.offsetMetadataManager.offset(
+ "foo",
+ "bar",
+ 1
+ ));
+ }
+
@Test
public void testOffsetCommitsSensor() {
OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java
index d768f4a4a7f..e0d4dd08cdf 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java
@@ -19,6 +19,8 @@ package org.apache.kafka.coordinator.group.runtime;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.NotCoordinatorException;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.coordinator.group.metrics.CoordinatorMetrics;
@@ -29,8 +31,11 @@ import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.image.MetadataProvenance;
import org.apache.kafka.server.util.timer.MockTimer;
import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
import org.apache.kafka.timeline.TimelineHashSet;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.ArgumentMatcher;
import java.time.Duration;
@@ -58,6 +63,7 @@ import static org.apache.kafka.coordinator.group.runtime.CoordinatorRuntime.Coor
import static org.apache.kafka.test.TestUtils.assertFutureThrows;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -167,14 +173,24 @@ public class CoordinatorRuntimeTest {
*/
private static class MockPartitionWriter extends InMemoryPartitionWriter<String> {
private final int maxRecordsInBatch;
+ private final boolean failEndMarker;
public MockPartitionWriter() {
- this(Integer.MAX_VALUE);
+ this(Integer.MAX_VALUE, false);
}
public MockPartitionWriter(int maxRecordsInBatch) {
+ this(maxRecordsInBatch, false);
+ }
+
+ public MockPartitionWriter(boolean failEndMarker) {
+ this(Integer.MAX_VALUE, failEndMarker);
+ }
+
+ public MockPartitionWriter(int maxRecordsInBatch, boolean failEndMarker) {
super(false);
this.maxRecordsInBatch = maxRecordsInBatch;
+ this.failEndMarker = failEndMarker;
}
@Override
@@ -206,20 +222,44 @@ public class CoordinatorRuntimeTest {
records.size(), maxRecordsInBatch));
}
}
+
+ @Override
+ public long appendEndTransactionMarker(
+ TopicPartition tp,
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result
+ ) throws KafkaException {
+ if (failEndMarker) {
+ throw new KafkaException("Can't write end marker.");
+ }
+ return super.appendEndTransactionMarker(
+ tp,
+ producerId,
+ producerEpoch,
+ coordinatorEpoch,
+ result
+ );
+ }
}
/**
* A simple Coordinator implementation that stores the records into a set.
*/
static class MockCoordinatorShard implements CoordinatorShard<String> {
+ private final SnapshotRegistry snapshotRegistry;
private final TimelineHashSet<String> records;
+ private final TimelineHashMap<Long, TimelineHashSet<String>> pendingRecords;
private final CoordinatorTimer<Void, String> timer;
MockCoordinatorShard(
SnapshotRegistry snapshotRegistry,
CoordinatorTimer<Void, String> timer
) {
+ this.snapshotRegistry = snapshotRegistry;
this.records = new TimelineHashSet<>(snapshotRegistry, 0);
+ this.pendingRecords = new TimelineHashMap<>(snapshotRegistry, 0);
this.timer = timer;
}
@@ -229,7 +269,34 @@ public class CoordinatorRuntimeTest {
short producerEpoch,
String record
) throws RuntimeException {
- records.add(record);
+ if (producerId == RecordBatch.NO_PRODUCER_ID) {
+ records.add(record);
+ } else {
+ pendingRecords
+ .computeIfAbsent(producerId, __ -> new TimelineHashSet<>(snapshotRegistry, 0))
+ .add(record);
+ }
+ }
+
+ @Override
+ public void replayEndTransactionMarker(
+ long producerId,
+ short producerEpoch,
+ TransactionResult result
+ ) throws RuntimeException {
+ if (result == TransactionResult.COMMIT) {
+ TimelineHashSet<String> pending = pendingRecords.remove(producerId);
+ if (pending == null) return;
+ records.addAll(pending);
+ } else {
+ pendingRecords.remove(producerId);
+ }
+ }
+
+ Set<String> pendingRecords(long producerId) {
+ TimelineHashSet<String> pending = pendingRecords.get(producerId);
+ if (pending == null) return Collections.emptySet();
+ return Collections.unmodifiableSet(new HashSet<>(pending));
}
Set<String> records() {
@@ -739,7 +806,10 @@ public class CoordinatorRuntimeTest {
// Records have been replayed to the coordinator.
assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().records());
// Records have been written to the log.
- assertEquals(Arrays.asList("record1", "record2"), writer.records(TP));
+ assertEquals(Arrays.asList(
+ InMemoryPartitionWriter.LogEntry.value("record1"),
+ InMemoryPartitionWriter.LogEntry.value("record2")
+ ), writer.entries(TP));
// Write #2.
CompletableFuture<String> write2 = runtime.scheduleWriteOperation("write#2", TP, DEFAULT_WRITE_TIMEOUT,
@@ -757,7 +827,11 @@ public class CoordinatorRuntimeTest {
// Records have been replayed to the coordinator.
assertEquals(mkSet("record1", "record2", "record3"), ctx.coordinator.coordinator().records());
// Records have been written to the log.
- assertEquals(Arrays.asList("record1", "record2", "record3"), writer.records(TP));
+ assertEquals(Arrays.asList(
+ InMemoryPartitionWriter.LogEntry.value("record1"),
+ InMemoryPartitionWriter.LogEntry.value("record2"),
+ InMemoryPartitionWriter.LogEntry.value("record3")
+ ), writer.entries(TP));
// Write #3 but without any records.
CompletableFuture<String> write3 = runtime.scheduleWriteOperation("write#3", TP, DEFAULT_WRITE_TIMEOUT,
@@ -771,7 +845,11 @@ public class CoordinatorRuntimeTest {
assertEquals(0L, ctx.coordinator.lastCommittedOffset());
assertEquals(Arrays.asList(0L, 2L, 3L), ctx.coordinator.snapshotRegistry().epochsList());
assertEquals(mkSet("record1", "record2", "record3"), ctx.coordinator.coordinator().records());
- assertEquals(Arrays.asList("record1", "record2", "record3"), writer.records(TP));
+ assertEquals(Arrays.asList(
+ InMemoryPartitionWriter.LogEntry.value("record1"),
+ InMemoryPartitionWriter.LogEntry.value("record2"),
+ InMemoryPartitionWriter.LogEntry.value("record3")
+ ), writer.entries(TP));
// Commit write #1.
writer.commit(TP, 2);
@@ -1073,6 +1151,326 @@ public class CoordinatorRuntimeTest {
);
}
+ @ParameterizedTest
+ @EnumSource(value = TransactionResult.class)
+ public void testScheduleTransactionCompletion(TransactionResult result) throws ExecutionException, InterruptedException, TimeoutException {
+ MockTimer timer = new MockTimer();
+ MockPartitionWriter writer = new MockPartitionWriter();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(GroupCoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(GroupCoordinatorMetrics.class))
+ .build();
+
+ // Schedule the loading.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Collections.singletonList(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Transactional write #1.
+ CompletableFuture<String> write1 = runtime.scheduleTransactionalWriteOperation(
+ "write#1",
+ TP,
+ "transactional-id",
+ 100L,
+ (short) 5,
+ DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1")
+ );
+
+ // Verify that the write is not committed yet.
+ assertFalse(write1.isDone());
+
+ // The last written offset is updated.
+ assertEquals(2L, ctx.coordinator.lastWrittenOffset());
+ // The last committed offset does not change.
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ // A new snapshot is created.
+ assertEquals(Arrays.asList(0L, 2L), ctx.coordinator.snapshotRegistry().epochsList());
+ // Records have been replayed to the coordinator. They are stored in
+ // the pending set for now.
+ assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().pendingRecords(
+ 100L
+ ));
+ // Records have been written to the log.
+ assertEquals(Arrays.asList(
+ InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record1"),
+ InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record2")
+ ), writer.entries(TP));
+
+ // Complete transaction #1.
+ CompletableFuture<Void> complete1 = runtime.scheduleTransactionCompletion(
+ "complete#1",
+ TP,
+ 100L,
+ (short) 5,
+ 10,
+ result,
+ DEFAULT_WRITE_TIMEOUT
+ );
+
+ // Verify that the completion is not committed yet.
+ assertFalse(complete1.isDone());
+
+ // The last written offset is updated.
+ assertEquals(3L, ctx.coordinator.lastWrittenOffset());
+ // The last committed offset does not change.
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ // A new snapshot is created.
+ assertEquals(Arrays.asList(0L, 2L, 3L), ctx.coordinator.snapshotRegistry().epochsList());
+ // Records have been replayed to the coordinator.
+ if (result == TransactionResult.COMMIT) {
+ // They are now in the records set if committed.
+ assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().records());
+ } else {
+ // Or they are gone if aborted.
+ assertEquals(Collections.emptySet(), ctx.coordinator.coordinator().records());
+ }
+
+ // Records have been written to the log.
+ assertEquals(Arrays.asList(
+ InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record1"),
+ InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record2"),
+ InMemoryPartitionWriter.LogEntry.control(100L, (short) 5, 10, result)
+ ), writer.entries(TP));
+
+ // Commit write #1.
+ writer.commit(TP, 2);
+
+ // The write is completed.
+ assertTrue(write1.isDone());
+ assertEquals("response1", write1.get(5, TimeUnit.SECONDS));
+
+ // Commit completion #1.
+ writer.commit(TP, 3);
+
+ // The transaction is completed.
+ assertTrue(complete1.isDone());
+ assertNull(complete1.get(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void testScheduleTransactionCompletionWhenWriteTimesOut() throws InterruptedException {
+ MockTimer timer = new MockTimer();
+ MockPartitionWriter writer = new MockPartitionWriter();
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(GroupCoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(GroupCoordinatorMetrics.class))
+ .build();
+
+ // Loads the coordinator.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(0, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Collections.singletonList(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Complete #1. We should get a TimeoutException because the HWM will not advance.
+ CompletableFuture<Void> timedOutCompletion = runtime.scheduleTransactionCompletion(
+ "complete#1",
+ TP,
+ 100L,
+ (short) 5,
+ 10,
+ TransactionResult.COMMIT,
+ Duration.ofMillis(3)
+ );
+
+ // Verify that the state has been updated.
+ assertEquals(1L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Arrays.asList(0L, 1L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Advance clock to timeout Complete #1.
+ timer.advanceClock(4);
+
+ assertFutureThrows(timedOutCompletion, org.apache.kafka.common.errors.TimeoutException.class);
+
+ // Verify that the state is still the same. We don't revert when the
+ // operation timeouts because the record has been written to the log.
+ assertEquals(1L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Arrays.asList(0L, 1L), ctx.coordinator.snapshotRegistry().epochsList());
+ }
+
+ @Test
+ public void testScheduleTransactionCompletionWhenWriteFails() {
+ MockTimer timer = new MockTimer();
+ // The partition writer accepts records but fails on markers.
+ MockPartitionWriter writer = new MockPartitionWriter(true);
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(GroupCoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(GroupCoordinatorMetrics.class))
+ .build();
+
+ // Loads the coordinator.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(0, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Collections.singletonList(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Write #1. It should succeed and be applied to the coordinator.
+ runtime.scheduleTransactionalWriteOperation(
+ "write#1",
+ TP,
+ "transactional-id",
+ 100L,
+ (short) 5,
+ DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1"));
+
+ // Verify that the state has been updated.
+ assertEquals(2L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Arrays.asList(0L, 2L), ctx.coordinator.snapshotRegistry().epochsList());
+ assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().pendingRecords(100L));
+ assertEquals(Collections.emptySet(), ctx.coordinator.coordinator().records());
+
+ // Complete transaction #1. It should fail.
+ CompletableFuture<Void> complete1 = runtime.scheduleTransactionCompletion(
+ "complete#1",
+ TP,
+ 100L,
+ (short) 5,
+ 10,
+ TransactionResult.COMMIT,
+ DEFAULT_WRITE_TIMEOUT
+ );
+ assertFutureThrows(complete1, KafkaException.class);
+
+ // Verify that the state has not changed.
+ assertEquals(2L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Arrays.asList(0L, 2L), ctx.coordinator.snapshotRegistry().epochsList());
+ assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().pendingRecords(100L));
+ assertEquals(Collections.emptySet(), ctx.coordinator.coordinator().records());
+ }
+
+ @Test
+ public void testScheduleTransactionCompletionWhenReplayFails() {
+ MockTimer timer = new MockTimer();
+ MockPartitionWriter writer = new MockPartitionWriter();
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT)
+ .withLoader(new MockCoordinatorLoader())
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(GroupCoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(GroupCoordinatorMetrics.class))
+ .build();
+
+ // Loads the coordinator.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(0L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Collections.singletonList(0L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Override the coordinator with a coordinator that throws
+ // an exception when replayEndTransactionMarker is called.
+ SnapshotRegistry snapshotRegistry = ctx.coordinator.snapshotRegistry();
+ ctx.coordinator = new SnapshottableCoordinator<>(
+ new LogContext(),
+ snapshotRegistry,
+ new MockCoordinatorShard(snapshotRegistry, ctx.timer) {
+ @Override
+ public void replayEndTransactionMarker(
+ long producerId,
+ short producerEpoch,
+ TransactionResult result
+ ) throws RuntimeException {
+ throw new IllegalArgumentException("error");
+ }
+ },
+ TP
+ );
+
+ // Write #1. It should succeed and be applied to the coordinator.
+ runtime.scheduleTransactionalWriteOperation(
+ "write#1",
+ TP,
+ "transactional-id",
+ 100L,
+ (short) 5,
+ DEFAULT_WRITE_TIMEOUT,
+ state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1"));
+
+ // Verify that the state has been updated.
+ assertEquals(2L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Arrays.asList(0L, 2L), ctx.coordinator.snapshotRegistry().epochsList());
+ assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().pendingRecords(100L));
+ assertEquals(Collections.emptySet(), ctx.coordinator.coordinator().records());
+ assertEquals(Arrays.asList(
+ InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record1"),
+ InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record2")
+ ), writer.entries(TP));
+
+ // Complete transaction #1. It should fail.
+ CompletableFuture<Void> complete1 = runtime.scheduleTransactionCompletion(
+ "complete#1",
+ TP,
+ 100L,
+ (short) 5,
+ 10,
+ TransactionResult.COMMIT,
+ DEFAULT_WRITE_TIMEOUT
+ );
+ assertFutureThrows(complete1, IllegalArgumentException.class);
+
+ // Verify that the state has not changed.
+ assertEquals(2L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(0L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Arrays.asList(0L, 2L), ctx.coordinator.snapshotRegistry().epochsList());
+ assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().pendingRecords(100L));
+ assertEquals(Collections.emptySet(), ctx.coordinator.coordinator().records());
+ assertEquals(Arrays.asList(
+ InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record1"),
+ InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record2")
+ ), writer.entries(TP));
+ }
+
@Test
public void testScheduleReadOp() throws ExecutionException, InterruptedException, TimeoutException {
MockTimer timer = new MockTimer();
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/InMemoryPartitionWriter.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/InMemoryPartitionWriter.java
index 759bbb2ced2..3cf8d480e46 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/InMemoryPartitionWriter.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/InMemoryPartitionWriter.java
@@ -18,13 +18,17 @@ package org.apache.kafka.coordinator.group.runtime;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.requests.TransactionResult;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
/**
* An in-memory partition writer.
@@ -33,10 +37,141 @@ import java.util.concurrent.locks.ReentrantLock;
*/
public class InMemoryPartitionWriter<T> implements PartitionWriter<T> {
+ public static class LogEntry {
+ public static <T> LogEntry value(T value) {
+ return new LogValue<>(value);
+ }
+
+ public static <T> LogEntry value(
+ long producerId,
+ short producerEpoch,
+ T value
+ ) {
+ return new LogValue<>(
+ producerId,
+ producerEpoch,
+ value
+ );
+ }
+
+ public static LogEntry control(
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result
+ ) {
+ return new LogControl(
+ producerId,
+ producerEpoch,
+ coordinatorEpoch,
+ result
+ );
+ }
+ }
+
+ public static class LogValue<T> extends LogEntry {
+ public final long producerId;
+ public final short producerEpoch;
+ public final T value;
+
+ private LogValue(
+ long producerId,
+ short producerEpoch,
+ T value
+ ) {
+ this.producerId = producerId;
+ this.producerEpoch = producerEpoch;
+ this.value = value;
+ }
+
+ private LogValue(T value) {
+ this(
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ value
+ );
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ LogValue<?> that = (LogValue<?>) o;
+
+ return Objects.equals(value, that.value);
+ }
+
+ @Override
+ public int hashCode() {
+ return value != null ? value.hashCode() : 0;
+ }
+
+ @Override
+ public String toString() {
+ return "BasicRecord(" +
+ "producerId=" + producerId +
+ ", producerEpoch=" + producerEpoch +
+ ", value=" + value +
+ ')';
+ }
+ }
+
+ public static class LogControl extends LogEntry {
+ public final long producerId;
+ public final short producerEpoch;
+ public final int coordinatorEpoch;
+ public final TransactionResult result;
+
+ private LogControl(
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result
+ ) {
+ this.producerId = producerId;
+ this.producerEpoch = producerEpoch;
+ this.coordinatorEpoch = coordinatorEpoch;
+ this.result = result;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ LogControl that = (LogControl) o;
+
+ if (producerId != that.producerId) return false;
+ if (producerEpoch != that.producerEpoch) return false;
+ if (coordinatorEpoch != that.coordinatorEpoch) return false;
+ return result == that.result;
+ }
+
+ @Override
+ public int hashCode() {
+ int result1 = (int) (producerId ^ (producerId >>> 32));
+ result1 = 31 * result1 + (int) producerEpoch;
+ result1 = 31 * result1 + coordinatorEpoch;
+ result1 = 31 * result1 + (result != null ? result.hashCode() : 0);
+ return result1;
+ }
+
+ @Override
+ public String toString() {
+ return "ControlRecord(" +
+ "producerId=" + producerId +
+ ", producerEpoch=" + producerEpoch +
+ ", coordinatorEpoch=" + coordinatorEpoch +
+ ", result=" + result +
+ ')';
+ }
+ }
+
private class PartitionState {
private ReentrantLock lock = new ReentrantLock();
private List<Listener> listeners = new ArrayList<>();
- private List<T> records = new ArrayList<>();
+ private List<LogEntry> entries = new ArrayList<>();
private long endOffset = 0L;
private long committedOffset = 0L;
}
@@ -93,7 +228,11 @@ public class InMemoryPartitionWriter<T> implements PartitionWriter<T> {
PartitionState state = partitionState(tp);
state.lock.lock();
try {
- state.records.addAll(records);
+ state.entries.addAll(records.stream().map(record -> new LogValue<T>(
+ producerId,
+ producerEpoch,
+ record
+ )).collect(Collectors.toList()));
state.endOffset += records.size();
if (autoCommit) commit(tp, state.endOffset);
return state.endOffset;
@@ -102,6 +241,31 @@ public class InMemoryPartitionWriter<T> implements PartitionWriter<T> {
}
}
+ @Override
+ public long appendEndTransactionMarker(
+ TopicPartition tp,
+ long producerId,
+ short producerEpoch,
+ int coordinatorEpoch,
+ TransactionResult result
+ ) throws KafkaException {
+ PartitionState state = partitionState(tp);
+ state.lock.lock();
+ try {
+ state.entries.add(new LogControl(
+ producerId,
+ producerEpoch,
+ coordinatorEpoch,
+ result
+ ));
+ state.endOffset += 1;
+ if (autoCommit) commit(tp, state.endOffset);
+ return state.endOffset;
+ } finally {
+ state.lock.unlock();
+ }
+ }
+
public void commit(
TopicPartition tp,
long offset
@@ -133,13 +297,13 @@ public class InMemoryPartitionWriter<T> implements PartitionWriter<T> {
}
}
- public List<T> records(
+ public List<LogEntry> entries(
TopicPartition tp
) {
PartitionState state = partitionState(tp);
state.lock.lock();
try {
- return Collections.unmodifiableList(state.records);
+ return Collections.unmodifiableList(state.entries);
} finally {
state.lock.unlock();
}