KAFKA-15653: Pass requestLocal as argument to callback so we use the correct one for the thread #14629
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -747,96 +747,9 @@ class ReplicaManager(val config: KafkaConfig, | |
| (verifiedEntries.toMap, unverifiedEntries.toMap, errorEntries.toMap) | ||
| } | ||
|
|
||
| if (notYetVerifiedEntriesPerPartition.isEmpty || addPartitionsToTxnManager.isEmpty) { | ||
| appendEntries(verifiedEntriesPerPartition, internalTopicsAllowed, origin, requiredAcks, verificationGuards.toMap, | ||
| errorsPerPartition, sTime, recordConversionStatsCallback, timeout, responseCallback, delayedProduceLock)(requestLocal)(Map.empty) | ||
| } else { | ||
| // For unverified entries, send a request to verify. When verified, the append process will proceed via the callback. | ||
| // We verify above that all partitions use the same producer ID. | ||
|
|
@@ -846,7 +759,8 @@ class ReplicaManager(val config: KafkaConfig, | |
| producerId = batchInfo.producerId, | ||
| producerEpoch = batchInfo.producerEpoch, | ||
| topicPartitions = notYetVerifiedEntriesPerPartition.keySet.toSeq, | ||
| callback = KafkaRequestHandler.wrap(appendEntries(entriesPerPartition, internalTopicsAllowed, origin, requiredAcks, verificationGuards.toMap, | ||
|
Contributor
There was a problem hiding this comment. This approach works. It's bit complicated since there are multiple callbacks (from
Member
Author
There was a problem hiding this comment. I'm not sure exactly how this would work. I suppose we could have a flag "verified" in some sort of context, but where is that context stored? Right now the request considers that some partitions are verified and some are not. (I'm not sure the producer actually sends more than one partition), but the code is currently generalizable to handle multiple partitions. If this is the case the context needs to store which partitions are verified and which are not. I agree that this is complicated, but I would need some time to think about how this would work. I'm also not sure the implications of going through the .handle of the request path -- it would be as if we received another request in the metrics etc.
Contributor
There was a problem hiding this comment. I think the way it's implemented right know is that we're just continuing processing from the current point after the network operation has completed, so as far as the "application" logic is concerned, the machinery to resume the execution on the request pool is hidden from the "application" logic. This allows to do an arbitrary number of non-blocking waits with minimal disruption of the application logic flow (the change is localized in the point of non-blocking wait).
Contributor
There was a problem hiding this comment. Thanks, Justine and Artem. Agreed that the alternative approach has its own drawback. So, we could punt on this. |
||
| errorsPerPartition, sTime, recordConversionStatsCallback, timeout, responseCallback, delayedProduceLock)(_)(_), requestLocal) | ||
| )) | ||
| } | ||
| } else { | ||
|
|
@@ -864,6 +778,107 @@ class ReplicaManager(val config: KafkaConfig, | |
| } | ||
| } | ||
|
|
||
| private def appendEntries(allEntries: Map[TopicPartition, MemoryRecords], | ||
|
Contributor
There was a problem hiding this comment. Could we add a comment that this method will potentially be called in a different request thread and it should avoid accessing any thread unsafe data structures? |
||
| internalTopicsAllowed: Boolean, | ||
| origin: AppendOrigin, | ||
| requiredAcks: Short, | ||
| verificationGuards: Map[TopicPartition, VerificationGuard], | ||
|
Member
There was a problem hiding this comment.
One option to prevent future bugs could be to pass a deep copy of these objects to the callback (instead of a copy of the reference). Another thing we can do is to restrict the number of parameters required by the callback. Having said that, have we already considered not having this as a callback and instead appending this synchronously? I understand that it has drawbacks since it is in critical produce path but maybe that would be acceptable given the complexity of ensuring thread safety here?
Member
Author
There was a problem hiding this comment. These maps are not mutable right?
Member
There was a problem hiding this comment.
Don't know. I haven't really checked. I wanted to ask you instead if you have verified that :)
Member
Author
There was a problem hiding this comment. Sorry for my phrasing. I was saying they are mutable. 😅
Member
There was a problem hiding this comment.
Did you mean to say immutable here? If they are mutable, then we need to ensure that they are not concurrently mutated in different threads.
Member
There was a problem hiding this comment. I agree with @junrao here.
However, it does not mean that it should be written to the log. The minimum is to guarantee the ordering of the records based on the state changes.
Contributor
There was a problem hiding this comment.
Is it true that the ordering of the log records is not important? It seems that you want the latest log to reflect the latest state, right?
Yes, the only thing is whether appendEntries can queue up the log record in the right order. I am not sure simply acquiring the group lock when asynchronously calling appendEntries achieves this.
Member
Author
There was a problem hiding this comment. My understanding from David is that the main thing we must guarantee is that we can't update the in memory-state again until the corresponding records are written to the log. From that, I thought that as long as those steps are synchronous (and I suppose we need a lock for now) that is ok. The ordering of the requests themselves is not guaranteed already.
Contributor
There was a problem hiding this comment.
It's true that there is not strong ordering guarantee among different clients. However, the ordering how the group coordinator serializes them in the log seems important. Consider an example. A group coordinator receives a joinGroup request from client 1 first and generates a group record rec1=GroupA(members: client1). It then receives joinGroup request from client 2 and generates a group record rec2=GroupA(members: client1, client2). If rec1 comes after rec2 in the log and the coordinator fails over, the new coordinator will restore the group state to include only client1 as its members and lose the valid member client2. Consider another example that involves consumer offsets. A group coordinator starts with a state in which client 1 owns partition 1. It receives an offset commit request from client 1 on partition 1. This is a valid request since client 1 owns partition 1. So, it generates rec3= GroupA(offsetForPartition1=100). A rebalance happens. The coordinator re-assigns partition 1 to client 2 and generates rec4= GroupA(partition1 owned by client 2). If rec3 comes after rec4 in the log and the coordinator fails over, when the new coordinator rebuilds its state, rec3 will look like invalid since it happens when client1 no longer owns partition 1.
Contributor
There was a problem hiding this comment. The serialization can be done asynchronously, without holding locks and relying on synchronous execution. It looks like it's should be that hard to support that in the new group coordinator #14705. |
||
| errorsPerPartition: Map[TopicPartition, Errors], | ||
| sTime: Long, | ||
| recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit, | ||
| timeout: Long, | ||
| responseCallback: Map[TopicPartition, PartitionResponse] => Unit, | ||
| delayedProduceLock: Option[Lock]) | ||
| (requestLocal: RequestLocal) | ||
|
jolshan marked this conversation as resolved.
Outdated
|
||
| (unverifiedEntries: Map[TopicPartition, Errors]): Unit = { | ||
| val verifiedEntries = | ||
| if (unverifiedEntries.isEmpty) | ||
| allEntries | ||
| else | ||
| allEntries.filter { case (tp, _) => | ||
| !unverifiedEntries.contains(tp) | ||
| } | ||
|
|
||
| val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed, | ||
| origin, verifiedEntries, requiredAcks, requestLocal, verificationGuards.toMap) | ||
| debug("Produce to local log in %d ms".format(time.milliseconds - sTime)) | ||
|
|
||
| val errorResults = (unverifiedEntries ++ errorsPerPartition).map { | ||
| case (topicPartition, error) => | ||
| // translate transaction coordinator errors to known producer response errors | ||
| val customException = | ||
| error match { | ||
| case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction")) | ||
| case Errors.CONCURRENT_TRANSACTIONS | | ||
| Errors.COORDINATOR_LOAD_IN_PROGRESS | | ||
| Errors.COORDINATOR_NOT_AVAILABLE | | ||
| Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException( | ||
| s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}")) | ||
| case _ => None | ||
| } | ||
| topicPartition -> LogAppendResult( | ||
| LogAppendInfo.UNKNOWN_LOG_APPEND_INFO, | ||
| Some(customException.getOrElse(error.exception)), | ||
| hasCustomErrorMessage = customException.isDefined | ||
| ) | ||
| } | ||
|
|
||
| val allResults = localProduceResults ++ errorResults | ||
| val produceStatus = allResults.map { case (topicPartition, result) => | ||
| topicPartition -> ProducePartitionStatus( | ||
| result.info.lastOffset + 1, // required offset | ||
| new PartitionResponse( | ||
| result.error, | ||
| result.info.firstOffset, | ||
| result.info.lastOffset, | ||
| result.info.logAppendTime, | ||
| result.info.logStartOffset, | ||
| result.info.recordErrors, | ||
| result.errorMessage | ||
| ) | ||
| ) // response status | ||
| } | ||
|
|
||
| actionQueue.add { | ||
| () => | ||
| allResults.foreach { case (topicPartition, result) => | ||
| val requestKey = TopicPartitionOperationKey(topicPartition) | ||
| result.info.leaderHwChange match { | ||
| case LeaderHwChange.INCREASED => | ||
| // some delayed operations may be unblocked after HW changed | ||
| delayedProducePurgatory.checkAndComplete(requestKey) | ||
| delayedFetchPurgatory.checkAndComplete(requestKey) | ||
| delayedDeleteRecordsPurgatory.checkAndComplete(requestKey) | ||
| case LeaderHwChange.SAME => | ||
| // probably unblock some follower fetch requests since log end offset has been updated | ||
| delayedFetchPurgatory.checkAndComplete(requestKey) | ||
| case LeaderHwChange.NONE => | ||
| // nothing | ||
| } | ||
| } | ||
| } | ||
|
|
||
| recordConversionStatsCallback(localProduceResults.map { case (k, v) => k -> v.info.recordConversionStats }) | ||
|
|
||
| if (delayedProduceRequestRequired(requiredAcks, allEntries, allResults)) { | ||
| // create delayed produce operation | ||
| val produceMetadata = ProduceMetadata(requiredAcks, produceStatus) | ||
| val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock) | ||
|
|
||
| // create a list of (topic, partition) pairs to use as keys for this delayed produce operation | ||
| val producerRequestKeys = allEntries.keys.map(TopicPartitionOperationKey(_)).toSeq | ||
|
|
||
| // try to complete the request immediately, otherwise put it into the purgatory | ||
| // this is because while the delayed produce operation is being created, new | ||
| // requests may arrive and hence make this operation completable. | ||
| delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys) | ||
| } else { | ||
| // we can respond immediately | ||
| val produceResponseStatus = produceStatus.map { case (k, status) => k -> status.responseStatus } | ||
| responseCallback(produceResponseStatus) | ||
| } | ||
| } | ||
|
|
||
| private def partitionEntriesForVerification(verificationGuards: mutable.Map[TopicPartition, VerificationGuard], | ||
| entriesPerPartition: Map[TopicPartition, MemoryRecords], | ||
| verifiedEntries: mutable.Map[TopicPartition, MemoryRecords], | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.