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 |
|---|---|---|
|
|
@@ -50,28 +50,34 @@ object KafkaRequestHandler { | |
| } | ||
|
|
||
| /** | ||
| * Creates a wrapped callback to be executed synchronously on the calling request thread or asynchronously | ||
| * on an arbitrary request thread. | ||
| * NOTE: this function must be originally called from a request thread. | ||
| * @param asyncCompletionCallback A callback method that we intend to call from the current thread or in another | ||
| * thread after an asynchronous action completes. The RequestLocal passed in must | ||
| * belong to the request handler thread that is executing the callback. | ||
| * @param requestLocal The RequestLocal for the current request handler thread in case we need to execute the callback | ||
| * function synchronously from the calling thread. | ||
| * @return Wrapped callback will either immediately execute `asyncCompletionCallback` or schedule it on an arbitrary request thread | ||
| * depending on where it is called | ||
| */ | ||
| def wrapAsyncCallback[T](asyncCompletionCallback: (RequestLocal, T) => Unit, requestLocal: RequestLocal): T => Unit = { | ||
| val requestChannel = threadRequestChannel.get() | ||
| val currentRequest = threadCurrentRequest.get() | ||
| if (requestChannel == null || currentRequest == null) { | ||
| if (!bypassThreadCheck) | ||
| throw new IllegalStateException("Attempted to reschedule to request handler thread from non-request handler thread.") | ||
| T => asyncCompletionCallback(requestLocal, T) | ||
| } else { | ||
| T => { | ||
| if (threadCurrentRequest.get() == currentRequest) { | ||
| // If the callback is actually executed on the same request thread, we can directly execute | ||
|
Contributor
There was a problem hiding this comment. The case when the callback is executed on the same request thread only happens when the txn coordinator is not available. Since it's a rare case, I am wondering if we really need to optimize for that. It seems that it's simpler to always enqueue the callback. Then, we could probably get rid of the ThreadLocal stuff.
Member
Author
There was a problem hiding this comment. Without this, there were a ton of failing tests when David implemented his change. We never really figured out the issue. We also have to consider the case where we bypass the thread check (ie tests) where we don't have the callback queue set up.
Contributor
There was a problem hiding this comment. @jolshan : I wasn't suggesting simply skip the test
So,
Member
Author
There was a problem hiding this comment. Right -- we had that before and it caused issues @dajac has more context. In addition, for tests, we would need to set up the request channel etc.
Contributor
There was a problem hiding this comment. Thanks, Justine. It seems that we can't easily get rid of the ThreadLocal stuff. So, we can leave this as it is.
With the current PR, it seems that it's possible for the callback to be called before the request handler thread finishes processing the request (that generates the callback), right? Is that causing any problem?
Member
Author
There was a problem hiding this comment. Sorry maybe I'm not explaining myself well. We saw problems before when the callback was scheduled before the request returned. We fixed it by adding this check. If we want to remove this check, we will need to do further investigation for that case.
Contributor
There was a problem hiding this comment. My understanding is that concurrency is not a problem in production logic (i.e. the fact that callback might actually execute at any time before or after or concurrently with the passing thread), but unit tests rely on deterministic order of execution and so adding concurrency here makes them "flaky".
Contributor
There was a problem hiding this comment. Thanks for the explanation, Justine and Artem. This sounds good then.
Member
There was a problem hiding this comment. Can we file a JIRA for investigating the alternative? I agree with @junrao that it would be much simpler to reason about if we always schedule it.
Member
Author
There was a problem hiding this comment. |
||
| // it without re-scheduling it. | ||
| asyncCompletionCallback(requestLocal, T) | ||
| } else { | ||
| // The requestChannel and request are captured in this lambda, so when it's executed on the callback thread | ||
| // we can re-schedule the original callback on a request thread and update the metrics accordingly. | ||
| requestChannel.sendCallbackRequest(RequestChannel.CallbackRequest(newRequestLocal => asyncCompletionCallback(newRequestLocal, T), currentRequest)) | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -132,7 +138,7 @@ class KafkaRequestHandler( | |
| } | ||
|
|
||
| threadCurrentRequest.set(originalRequest) | ||
| callback.fun(requestLocal) | ||
| } catch { | ||
| case e: FatalExitError => | ||
| completeShutdown() | ||
|
|
@@ -174,6 +180,7 @@ class KafkaRequestHandler( | |
|
|
||
| private def completeShutdown(): Unit = { | ||
| requestLocal.close() | ||
| threadRequestChannel.remove() | ||
| shutdownComplete.countDown() | ||
| } | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -733,7 +733,6 @@ class ReplicaManager(val config: KafkaConfig, | |
| transactionalId: String = null, | ||
| actionQueue: ActionQueue = this.actionQueue): Unit = { | ||
| if (isValidRequiredAcks(requiredAcks)) { | ||
|
|
||
| val verificationGuards: mutable.Map[TopicPartition, VerificationGuard] = mutable.Map[TopicPartition, VerificationGuard]() | ||
| val (verifiedEntriesPerPartition, notYetVerifiedEntriesPerPartition, errorsPerPartition) = | ||
|
|
@@ -747,96 +746,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, 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 +758,20 @@ class ReplicaManager(val config: KafkaConfig, | |
| producerId = batchInfo.producerId, | ||
| producerEpoch = batchInfo.producerEpoch, | ||
| topicPartitions = notYetVerifiedEntriesPerPartition.keySet.toSeq, | ||
| callback = KafkaRequestHandler.wrapAsyncCallback( | ||
| appendEntries( | ||
| entriesPerPartition, | ||
| internalTopicsAllowed, | ||
| origin, | ||
| requiredAcks, | ||
| verificationGuards.toMap, | ||
| errorsPerPartition, | ||
| recordConversionStatsCallback, | ||
| timeout, | ||
| responseCallback, | ||
| delayedProduceLock | ||
| ), | ||
| requestLocal) | ||
| )) | ||
| } | ||
| } else { | ||
|
|
@@ -864,6 +789,110 @@ class ReplicaManager(val config: KafkaConfig, | |
| } | ||
| } | ||
|
|
||
| /* | ||
| * Note: This method can be used as a callback in a different request thread. Ensure that correct RequestLocal | ||
| * is passed when executing this method. Accessing non-thread-safe data structures should be avoided if possible. | ||
| */ | ||
| 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], | ||
| recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit, | ||
| timeout: Long, | ||
| responseCallback: Map[TopicPartition, PartitionResponse] => Unit, | ||
| delayedProduceLock: Option[Lock]) | ||
| (requestLocal: RequestLocal, unverifiedEntries: Map[TopicPartition, Errors]): Unit = { | ||
| val sTime = time.milliseconds | ||
| 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], | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ismael mentioned in #9229 (comment) that thread locals are most useful when one doesn't control the code. So, I am wondering if we could get rid of the two ThreadLocal:
threadRequestChannelandthreadCurrentRequestinKafkaRequestHandlerintroduced in KAFKA-14561. The reason for the former is to obtain requestChannel. We could simply pass in requestChannel toReplicaManager.appendRecords. The reason for the latter is (1) to obtain currentRequest and (2) to make sure that the callback can be short-circuited if it's called on the same request handler thread. For (1), we could also pass in currentRequest toReplicaManager.appendRecords. For (2), we could change the code such thatKafkaRequestHandler.wrapis called only when the callback truly needs to be run from a different thread. Otherwise, we can just call the callback directly there.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmmm. I'm not sure it makes sense to try to pass the request channel and current in to every method we want to do a callback for.
As for 2 and short circuiting... Are you suggesting I move the wrap call into AddPartitionsToTxnManager? If so, I need to pass the appendEntries method in there as well. That may be trickier with the typing.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Currently, the callback is only needed for produce request in ReplicaManager. If you look at
KafkaApis.handleProduceRequest, we already pass in both request channel and current request toReplicaManager.appendRecordsthroughsendResponseCallback.Yes. Currently, we already pass in
appendEntriesas a callback toAddPartitionsToTxnManager, right?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We pass the wrapped method which is a simpler type.
type AppendCallback = Map[TopicPartition, Errors] => UnitThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
As for just produce. Yes, this is the case now, but I think Artem was trying to create the wrap method as a "general" callback mechanism.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks, Justine and Artem. Yes, passing around the context has it's own issues. The main thing with thread local is to make sure that we don't introduce any GC issues. Currently, it seems that we never remove
threadRequestChannel. Since we allow dynamically changing the number of request handler threads, it's probably better to removethreadRequestChannelwhen the thread completes?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm. We assign the request channel when the thread is created. I assume that request channel will remain with the thread during its lifetime. Is that incorrect?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
True in the common case. I was wondering what happens if we dynamically reduce the number of request handler threads.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If we reduce then I assume the thread just stops?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I understand the issue. I have a commit below that clears the request local.