Skip to content

KAFKA-10080; Fix race condition on txn completion which can cause duplicate appends #8782

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 3 commits into from
Jun 3, 2020
Merged

KAFKA-10080; Fix race condition on txn completion which can cause duplicate appends #8782

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
44d2815
KAFKA-10080; Fix race condition on txn completion which can cause dup…
hachikuji Jun 2, 2020
9d1d560
Rename pendingCommitTxn to pendingCompleteTxn
hachikuji Jun 2, 2020
c52dc75
Fix failing test due to nice mock change
hachikuji Jun 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 19 additions & 24 deletions core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,13 +214,11 @@ class TransactionMarkerChannelManager(config: KafkaConfig,
}
}

private def writeTxnCompletion(pendingCommitTxn: PendingCompleteTxn): Unit = {
transactionsWithPendingMarkers.remove(pendingCommitTxn.transactionalId)

val transactionalId = pendingCommitTxn.transactionalId
val txnMetadata = pendingCommitTxn.txnMetadata
val newMetadata = pendingCommitTxn.newMetadata
val coordinatorEpoch = pendingCommitTxn.coordinatorEpoch
private def writeTxnCompletion(pendingCompleteTxn: PendingCompleteTxn): Unit = {
val transactionalId = pendingCompleteTxn.transactionalId
val txnMetadata = pendingCompleteTxn.txnMetadata
val newMetadata = pendingCompleteTxn.newMetadata
val coordinatorEpoch = pendingCompleteTxn.coordinatorEpoch

trace(s"Completed sending transaction markers for $transactionalId; begin transition " +
s"to ${newMetadata.txnState}")
Expand All @@ -242,7 +240,6 @@ class TransactionMarkerChannelManager(config: KafkaConfig,
if (epochAndMetadata.coordinatorEpoch == coordinatorEpoch) {
debug(s"Sending $transactionalId's transaction markers for $txnMetadata with " +
s"coordinator epoch $coordinatorEpoch succeeded, trying to append complete transaction log now")

tryAppendToLog(PendingCompleteTxn(transactionalId, coordinatorEpoch, txnMetadata, newMetadata))
} else {
info(s"The cached metadata $txnMetadata has changed to $epochAndMetadata after " +
Expand All @@ -263,15 +260,13 @@ class TransactionMarkerChannelManager(config: KafkaConfig,
txnMetadata: TransactionMetadata,
newMetadata: TxnTransitMetadata): Unit = {
val transactionalId = txnMetadata.transactionalId

val pendingCommitTxn = PendingCompleteTxn(
val pendingCompleteTxn = PendingCompleteTxn(
transactionalId,
coordinatorEpoch,
txnMetadata,
newMetadata
)
newMetadata)

transactionsWithPendingMarkers.put(transactionalId, pendingCommitTxn)
transactionsWithPendingMarkers.put(transactionalId, pendingCompleteTxn)
addTxnMarkersToBrokerQueue(transactionalId, txnMetadata.producerId,
txnMetadata.producerEpoch, txnResult, coordinatorEpoch, txnMetadata.topicPartitions.toSet)
maybeWriteTxnCompletion(transactionalId)
Expand All @@ -285,15 +280,16 @@ class TransactionMarkerChannelManager(config: KafkaConfig,
}
}

private def maybeWriteTxnCompletion(transactionalId: String): Unit = {
Option(transactionsWithPendingMarkers.get(transactionalId)).foreach { pendingCommitTxn =>

@hachikuji hachikuji Jun 2, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Multiple threads may see the transaction still as pending and attempt completion.

if (!hasPendingMarkersToWrite(pendingCommitTxn.txnMetadata)) {
writeTxnCompletion(pendingCommitTxn)
def maybeWriteTxnCompletion(transactionalId: String): Unit = {
Option(transactionsWithPendingMarkers.get(transactionalId)).foreach { pendingCompleteTxn =>
if (!hasPendingMarkersToWrite(pendingCompleteTxn.txnMetadata) &&
transactionsWithPendingMarkers.remove(transactionalId, pendingCompleteTxn)) {
writeTxnCompletion(pendingCompleteTxn)
}
}
}

private def tryAppendToLog(txnLogAppend: PendingCompleteTxn) = {
private def tryAppendToLog(txnLogAppend: PendingCompleteTxn): Unit = {
// try to append to the transaction log
def appendCallback(error: Errors): Unit =
error match {
Expand Down Expand Up @@ -404,18 +400,17 @@ class TransactionMarkerChannelManager(config: KafkaConfig,
def removeMarkersForTxnId(transactionalId: String): Unit = {
transactionsWithPendingMarkers.remove(transactionalId)
}

def completeSendMarkersForTxnId(transactionalId: String): Unit = {
maybeWriteTxnCompletion(transactionalId)
}
}

case class TxnIdAndMarkerEntry(txnId: String, txnMarkerEntry: TxnMarkerEntry)

case class PendingCompleteTxn(transactionalId: String, coordinatorEpoch: Int, txnMetadata: TransactionMetadata, newMetadata: TxnTransitMetadata) {
case class PendingCompleteTxn(transactionalId: String,
coordinatorEpoch: Int,
txnMetadata: TransactionMetadata,
newMetadata: TxnTransitMetadata) {

override def toString: String = {
"TxnLogAppend(" +
"PendingCompleteTxn(" +
s"transactionalId=$transactionalId, " +
s"coordinatorEpoch=$coordinatorEpoch, " +
s"txnMetadata=$txnMetadata, " +
Expand Down
2 changes: 1 addition & 1 deletion .../main/scala/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,7 @@ class TransactionMarkerRequestCompletionHandler(brokerId: Int,
txnMarker.coordinatorEpoch,
retryPartitions.toSet)
} else {
txnMarkerChannelManager.completeSendMarkersForTxnId(transactionalId)
txnMarkerChannelManager.maybeWriteTxnCompletion(transactionalId)
}
}
}
Expand Down
87 changes: 77 additions & 10 deletions ...c/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,10 @@
*/
package kafka.coordinator.transaction

import java.util
import java.util.Arrays.asList
import java.util.Collections
import java.util.concurrent.{Callable, Executors, Future}

import kafka.common.RequestAndCompletionHandler
import kafka.metrics.KafkaYammerMetrics
Expand All @@ -34,11 +37,12 @@ import org.junit.Test

import scala.jdk.CollectionConverters._
import scala.collection.mutable
import scala.util.Try

class TransactionMarkerChannelManagerTest {
private val metadataCache: MetadataCache = EasyMock.createNiceMock(classOf[MetadataCache])
private val networkClient: NetworkClient = EasyMock.createNiceMock(classOf[NetworkClient])
private val txnStateManager: TransactionStateManager = EasyMock.createNiceMock(classOf[TransactionStateManager])
private val txnStateManager: TransactionStateManager = EasyMock.mock(classOf[TransactionStateManager])

private val partition1 = new TopicPartition("topic1", 0)
private val partition2 = new TopicPartition("topic1", 1)
Expand Down Expand Up @@ -86,6 +90,70 @@ class TransactionMarkerChannelManagerTest {
.anyTimes()
}

@Test
def shouldOnlyWriteTxnCompletionOnce(): Unit = {

@guozhangwang guozhangwang Jun 2, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this test cover concurrent calls to maybeWriteTxnCompletion?

@hachikuji hachikuji Jun 2, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It does. I was trying to setup this test to fit how we're likely hitting this in practice. In the call to addTxnMarkersToSend, before calling maybeWriteTxnCompletion, we have to acquire the lock. It is possible that the caller fails to acquire the lock before the markers finish getting written and the transaction gets completed in the request completion handler.

@guozhangwang guozhangwang Jun 3, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, so when the bug still exist this test would probably not fail consistently, but would be flaky, right?

@hachikuji hachikuji Jun 3, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It fails deterministically without the fix.

@guozhangwang guozhangwang Jun 3, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah yes, I missed the txnMetadata2.lock.lock() before starting the scheduler, thanks.

mockCache()

val expectedTransition = txnMetadata2.prepareComplete(time.milliseconds())

EasyMock.expect(metadataCache.getPartitionLeaderEndpoint(
EasyMock.eq(partition1.topic),
EasyMock.eq(partition1.partition),
EasyMock.anyObject())
).andReturn(Some(broker1)).anyTimes()

EasyMock.expect(txnStateManager.appendTransactionToLog(
EasyMock.eq(transactionalId2),
EasyMock.eq(coordinatorEpoch),
EasyMock.eq(expectedTransition),
EasyMock.capture(capturedErrorsCallback),
EasyMock.anyObject()))
.andAnswer(() => {
txnMetadata2.completeTransitionTo(expectedTransition)
capturedErrorsCallback.getValue.apply(Errors.NONE)
}).once()

EasyMock.replay(txnStateManager, metadataCache)

var addMarkerFuture: Future[Try[Unit]] = null
val executor = Executors.newFixedThreadPool(1)
txnMetadata2.lock.lock()
try {
addMarkerFuture = executor.submit((() => {
Try(channelManager.addTxnMarkersToSend(coordinatorEpoch, txnResult,
txnMetadata2, expectedTransition))
}): Callable[Try[Unit]])

val header = new RequestHeader(ApiKeys.WRITE_TXN_MARKERS, 0, "client", 1)
val response = new WriteTxnMarkersResponse(
Collections.singletonMap(producerId2: java.lang.Long, Collections.singletonMap(partition1, Errors.NONE)))
val clientResponse = new ClientResponse(header, null, null,
time.milliseconds(), time.milliseconds(), false, null, null,
response)

TestUtils.waitUntilTrue(() => {
val requests = channelManager.drainQueuedTransactionMarkers()
if (requests.nonEmpty) {
assertEquals(1, requests.size)
val request = requests.head
request.handler.onComplete(clientResponse)
true
} else {
false
}
}, "Timed out waiting for expected WriteTxnMarkers request")
} finally {
txnMetadata2.lock.unlock()
executor.shutdown()
}

assertNotNull(addMarkerFuture)
assertTrue("Add marker task failed with exception " + addMarkerFuture.get().get,
addMarkerFuture.get().isSuccess)

EasyMock.verify(txnStateManager)
}

@Test
def shouldGenerateEmptyMapWhenNoRequestsOutstanding(): Unit = {
assertTrue(channelManager.generateRequests().isEmpty)
Expand Down Expand Up @@ -153,7 +221,6 @@ class TransactionMarkerChannelManagerTest {
EasyMock.replay(metadataCache)

channelManager.addTxnMarkersToSend(coordinatorEpoch, txnResult, txnMetadata1, txnMetadata1.prepareComplete(time.milliseconds()))
channelManager.addTxnMarkersToSend(coordinatorEpoch, txnResult, txnMetadata2, txnMetadata2.prepareComplete(time.milliseconds()))

@hachikuji hachikuji Jun 2, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The change to use mock instead of niceMock led to a failure because of an unexpected append to the log. It seemed like this call was not necessary to test the behavior we were interested in here, so I removed it rather than adding the expected call to append.


assertEquals(1, channelManager.numTxnsWithPendingMarkers)
assertEquals(1, channelManager.queueForBroker(broker2.id).get.totalNumMarkers)
Expand Down Expand Up @@ -291,7 +358,7 @@ class TransactionMarkerChannelManagerTest {

val response = new WriteTxnMarkersResponse(createPidErrorMap(Errors.NONE))
for (requestAndHandler <- requestAndHandlers) {
requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1),
requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.WRITE_TXN_MARKERS, 0, "client", 1),

@guozhangwang guozhangwang Jun 2, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice catch. Not sure why they did not fail before :)

null, null, 0, 0, false, null, null, response))
}

Expand Down Expand Up @@ -338,7 +405,7 @@ class TransactionMarkerChannelManagerTest {

val response = new WriteTxnMarkersResponse(createPidErrorMap(Errors.NONE))
for (requestAndHandler <- requestAndHandlers) {
requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1),
requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.WRITE_TXN_MARKERS, 0, "client", 1),
null, null, 0, 0, false, null, null, response))
}

Expand Down Expand Up @@ -387,7 +454,7 @@ class TransactionMarkerChannelManagerTest {

val response = new WriteTxnMarkersResponse(createPidErrorMap(Errors.NONE))
for (requestAndHandler <- requestAndHandlers) {
requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1),
requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.WRITE_TXN_MARKERS, 0, "client", 1),
null, null, 0, 0, false, null, null, response))
}

Expand All @@ -402,7 +469,7 @@ class TransactionMarkerChannelManagerTest {
assertEquals(CompleteCommit, txnMetadata2.state)
}

private def createPidErrorMap(errors: Errors) = {
private def createPidErrorMap(errors: Errors): util.HashMap[java.lang.Long, util.Map[TopicPartition, Errors]] = {
val pidMap = new java.util.HashMap[java.lang.Long, java.util.Map[TopicPartition, Errors]]()
val errorsMap = new java.util.HashMap[TopicPartition, Errors]()
errorsMap.put(partition1, errors)
Expand All @@ -414,11 +481,11 @@ class TransactionMarkerChannelManagerTest {
def shouldCreateMetricsOnStarting(): Unit = {
val metrics = KafkaYammerMetrics.defaultRegistry.allMetrics.asScala

assertEquals(1, metrics.filter { case (k, _) =>
assertEquals(1, metrics.count { case (k, _) =>
k.getMBeanName == "kafka.coordinator.transaction:type=TransactionMarkerChannelManager,name=UnknownDestinationQueueSize"
}.size)
assertEquals(1, metrics.filter { case (k, _) =>
})
assertEquals(1, metrics.count { case (k, _) =>
k.getMBeanName == "kafka.coordinator.transaction:type=TransactionMarkerChannelManager,name=LogAppendRetryQueueSize"
}.size)
})
}
}
2 changes: 1 addition & 1 deletion ...la/unit/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandlerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,7 @@ class TransactionMarkerRequestCompletionHandlerTest {
private def verifyCompleteDelayedOperationOnError(error: Errors): Unit = {

var completed = false
EasyMock.expect(markerChannelManager.completeSendMarkersForTxnId(transactionalId))
EasyMock.expect(markerChannelManager.maybeWriteTxnCompletion(transactionalId))
.andAnswer(() => completed = true)
.once()
EasyMock.replay(markerChannelManager)
Expand Down