You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dajac (via GitHub)" <gi...@apache.org> on 2024/01/11 13:08:55 UTC
[PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
dajac opened a new pull request, #15176:
URL: https://github.com/apache/kafka/pull/15176
This patch removes the extra hop via the request thread when the new group coordinator verifies a transaction. Prior to it, the ReplicaManager would automatically re-schedule the callback to a request thread. However, the new group coordinator does not need this as it already schedules the write into its own thread.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1454268198
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1090,38 +1090,29 @@ class ReplicaManager(val config: KafkaConfig,
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
* @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the error if it exists or Errors.NONE.
* If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned.
- * This guard can not be used for verification and any appends that attenpt to use it will fail.
+ * This guard can not be used for verification and any appends that attempt to use it will fail.
*/
def maybeStartTransactionVerificationForPartition(
topicPartition: TopicPartition,
transactionalId: String,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: Either[Errors, VerificationGuard] => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
- preAppendErrors.getOrElse(topicPartition, Errors.NONE),
- newRequestLocal,
- verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
+ def generalizedCallback(results: Map[TopicPartition, Either[Errors, VerificationGuard]]): Unit = {
Review Comment:
https://github.com/apache/kafka/pull/15087
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1470109774
##########
core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorConcurrencyTest.scala:
##########
@@ -85,6 +85,10 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest
groupCoordinator = GroupCoordinator(config, replicaManager, heartbeatPurgatory, rebalancePurgatory, timer.time, metrics)
groupCoordinator.startup(() => zkClient.getTopicPartitionCount(Topic.GROUP_METADATA_TOPIC_NAME).getOrElse(config.offsetsTopicPartitions),
false)
+
+ // Transactional appends attempt to schedule to the request handler thread using
+ // a non request handler thread. Set this to avoid error.
+ KafkaRequestHandler.setBypassThreadCheck(true)
Review Comment:
That’s right.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1451074358
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1090,38 +1090,29 @@ class ReplicaManager(val config: KafkaConfig,
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
* @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the error if it exists or Errors.NONE.
* If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned.
- * This guard can not be used for verification and any appends that attenpt to use it will fail.
+ * This guard can not be used for verification and any appends that attempt to use it will fail.
*/
def maybeStartTransactionVerificationForPartition(
topicPartition: TopicPartition,
transactionalId: String,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: Either[Errors, VerificationGuard] => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
- preAppendErrors.getOrElse(topicPartition, Errors.NONE),
- newRequestLocal,
- verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
+ def generalizedCallback(results: Map[TopicPartition, Either[Errors, VerificationGuard]]): Unit = {
+ callback(results.getOrElse(topicPartition, Right(VerificationGuard.SENTINEL)))
Review Comment:
This logic is just a translation from the current implementation (so it's not introducing anything new), but is it expected that we don't get the results for the requested topicPartition? Should we log a warning, so that we know that we're hitting some unexpected code path?
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1090,38 +1090,29 @@ class ReplicaManager(val config: KafkaConfig,
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
* @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the error if it exists or Errors.NONE.
* If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned.
- * This guard can not be used for verification and any appends that attenpt to use it will fail.
+ * This guard can not be used for verification and any appends that attempt to use it will fail.
*/
def maybeStartTransactionVerificationForPartition(
topicPartition: TopicPartition,
transactionalId: String,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: Either[Errors, VerificationGuard] => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
- preAppendErrors.getOrElse(topicPartition, Errors.NONE),
- newRequestLocal,
- verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
+ def generalizedCallback(results: Map[TopicPartition, Either[Errors, VerificationGuard]]): Unit = {
Review Comment:
I think we could do the translation from preAppendErrors, newRequestLocal, verificationGuards here, then we'd avoid propagating the changes all the way to replication layer.
##########
core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala:
##########
@@ -935,8 +935,9 @@ private[group] class GroupCoordinator(
producerId,
producerEpoch,
RecordBatch.NO_SEQUENCE,
- requestLocal,
- postVerificationCallback
+ // Wrap the callback to be handled on an arbitrary request handler thread
+ // when transaction verification is complete.
+ KafkaRequestHandler.wrapAsyncCallback(postVerificationCallback, requestLocal)
Review Comment:
It's interesting to note (I don't think we need to change anything) is that now we'll have a production code path (and not just unit test) where we can call wrapped callback on the same request thread and we'll go through the optimized code path where we call the callback directly.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1132,36 +1123,31 @@ class ReplicaManager(val config: KafkaConfig,
* @param transactionalId the transactional id for the transaction
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the errors per topic partition if there were errors.
* The callback will also be supplied the verification guards per partition if they exist. It is possible to have an
* error and a verification guard for a topic partition if the topic partition was unable to be verified by the transaction
- * coordinator. Transaction coordinator errors are mapped to append-friendly errors. The callback is wrapped so that it
- * is scheduled on a request handler thread. There, it should be called with that request handler thread's thread local and
- * not the one supplied to this method.
+ * coordinator. Transaction coordinator errors are mapped to append-friendly errors.
*/
def maybeStartTransactionVerificationForPartitions(
topicPartitionBatchInfo: Map[TopicPartition, Int],
transactionalId: String,
producerId: Long,
producerEpoch: Short,
- requestLocal: RequestLocal,
- callback: (Map[TopicPartition, Errors], RequestLocal, Map[TopicPartition, VerificationGuard]) => Unit
+ callback: mutable.Map[TopicPartition, Either[Errors, VerificationGuard]] => Unit
Review Comment:
My understanding, that once we refactor these changes, this function could be either called from GC code path (that may not care about requestLocal) or from the core data path, that needs requestLocal, because the callback may be called immediately in this thread context.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #15176:
URL: https://github.com/apache/kafka/pull/15176#issuecomment-1916310489
Thanks @jolshan. I have addressed your comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1456228457
##########
core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala:
##########
@@ -935,8 +935,9 @@ private[group] class GroupCoordinator(
producerId,
producerEpoch,
RecordBatch.NO_SEQUENCE,
- requestLocal,
- postVerificationCallback
+ // Wrap the callback to be handled on an arbitrary request handler thread
+ // when transaction verification is complete.
+ KafkaRequestHandler.wrapAsyncCallback(postVerificationCallback, requestLocal)
Review Comment:
Oh sorry. I guess I was just confused I didn't see it in the replica manager flow. (for produce)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1470117099
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -813,19 +816,19 @@ class ReplicaManager(val config: KafkaConfig,
val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
- entriesPerPartition.foreach { case (topicPartition, records) =>
+ entriesPerPartition.forKeyValue { (topicPartition, records) =>
// Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
transactionalBatches.foreach(batch => transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
- if (!transactionalBatches.isEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
+ if (transactionalBatches.nonEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
}
if (transactionalProducerInfo.size > 1) {
throw new InvalidPidMappingException("Transactional records contained more than one producer ID")
}
- def postVerificationCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
+ def postVerificationCallback(newRequestLocal: RequestLocal,
+ results: (Map[TopicPartition, Errors], Map[TopicPartition, VerificationGuard])): Unit = {
Review Comment:
Ah right -- makes sense 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1454771868
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1090,38 +1090,29 @@ class ReplicaManager(val config: KafkaConfig,
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
* @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the error if it exists or Errors.NONE.
* If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned.
- * This guard can not be used for verification and any appends that attenpt to use it will fail.
+ * This guard can not be used for verification and any appends that attempt to use it will fail.
*/
def maybeStartTransactionVerificationForPartition(
topicPartition: TopicPartition,
transactionalId: String,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: Either[Errors, VerificationGuard] => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
- preAppendErrors.getOrElse(topicPartition, Errors.NONE),
- newRequestLocal,
- verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
+ def generalizedCallback(results: Map[TopicPartition, Either[Errors, VerificationGuard]]): Unit = {
Review Comment:
Ack. I will take a look at https://github.com/apache/kafka/pull/15087.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1454273994
##########
core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala:
##########
@@ -935,8 +935,9 @@ private[group] class GroupCoordinator(
producerId,
producerEpoch,
RecordBatch.NO_SEQUENCE,
- requestLocal,
- postVerificationCallback
+ // Wrap the callback to be handled on an arbitrary request handler thread
+ // when transaction verification is complete.
+ KafkaRequestHandler.wrapAsyncCallback(postVerificationCallback, requestLocal)
Review Comment:
Why are we calling this here? I thought we wanted to avoid this wrap here and only do it for produce requests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1453039755
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1132,36 +1123,31 @@ class ReplicaManager(val config: KafkaConfig,
* @param transactionalId the transactional id for the transaction
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the errors per topic partition if there were errors.
* The callback will also be supplied the verification guards per partition if they exist. It is possible to have an
* error and a verification guard for a topic partition if the topic partition was unable to be verified by the transaction
- * coordinator. Transaction coordinator errors are mapped to append-friendly errors. The callback is wrapped so that it
- * is scheduled on a request handler thread. There, it should be called with that request handler thread's thread local and
- * not the one supplied to this method.
+ * coordinator. Transaction coordinator errors are mapped to append-friendly errors.
*/
def maybeStartTransactionVerificationForPartitions(
topicPartitionBatchInfo: Map[TopicPartition, Int],
transactionalId: String,
producerId: Long,
producerEpoch: Short,
- requestLocal: RequestLocal,
- callback: (Map[TopicPartition, Errors], RequestLocal, Map[TopicPartition, VerificationGuard]) => Unit
+ callback: mutable.Map[TopicPartition, Either[Errors, VerificationGuard]] => Unit
Review Comment:
This is correct. The `requestLocal` is not required within this function though. The caller if it uses `wrapAsyncCallback` will get the correct one to use.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #15176:
URL: https://github.com/apache/kafka/pull/15176#issuecomment-1887429299
@artemlivshits @jolshan Could you please take a look at this one when you get a chance? Let me know what you think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1453051782
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1090,38 +1090,29 @@ class ReplicaManager(val config: KafkaConfig,
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
* @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the error if it exists or Errors.NONE.
* If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned.
- * This guard can not be used for verification and any appends that attenpt to use it will fail.
+ * This guard can not be used for verification and any appends that attempt to use it will fail.
*/
def maybeStartTransactionVerificationForPartition(
topicPartition: TopicPartition,
transactionalId: String,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: Either[Errors, VerificationGuard] => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
- preAppendErrors.getOrElse(topicPartition, Errors.NONE),
- newRequestLocal,
- verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
+ def generalizedCallback(results: Map[TopicPartition, Either[Errors, VerificationGuard]]): Unit = {
+ callback(results.getOrElse(topicPartition, Right(VerificationGuard.SENTINEL)))
Review Comment:
It should not happen but who knows. Let me log an error and fail the callback if it would ever happen.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1453042161
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1090,38 +1090,29 @@ class ReplicaManager(val config: KafkaConfig,
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
* @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the error if it exists or Errors.NONE.
* If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned.
- * This guard can not be used for verification and any appends that attenpt to use it will fail.
+ * This guard can not be used for verification and any appends that attempt to use it will fail.
*/
def maybeStartTransactionVerificationForPartition(
topicPartition: TopicPartition,
transactionalId: String,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: Either[Errors, VerificationGuard] => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
- preAppendErrors.getOrElse(topicPartition, Errors.NONE),
- newRequestLocal,
- verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
+ def generalizedCallback(results: Map[TopicPartition, Either[Errors, VerificationGuard]]): Unit = {
Review Comment:
Yeah, I was debating about this. The reasoning for not doing it here is that @jolshan may also use `maybeStartTransactionVerificationForPartitions` from the core path in conjunction with `wrapAsyncCallback` so she will also need to have a unary callback. @jolshan Is my understanding correct? Otherwise, we could limit the change to this method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1454267538
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1090,38 +1090,29 @@ class ReplicaManager(val config: KafkaConfig,
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
* @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the error if it exists or Errors.NONE.
* If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned.
- * This guard can not be used for verification and any appends that attenpt to use it will fail.
+ * This guard can not be used for verification and any appends that attempt to use it will fail.
*/
def maybeStartTransactionVerificationForPartition(
topicPartition: TopicPartition,
transactionalId: String,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: Either[Errors, VerificationGuard] => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
- preAppendErrors.getOrElse(topicPartition, Errors.NONE),
- newRequestLocal,
- verificationGuards.getOrElse(topicPartition, VerificationGuard.SENTINEL))
+ def generalizedCallback(results: Map[TopicPartition, Either[Errors, VerificationGuard]]): Unit = {
Review Comment:
Please take a look at my refactor PR. I have some this to some extent.
I'd prefer not to overhaul it again (as I did after the previous group coordinator change)
Hopefully it makes this work easier too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1470109481
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -813,19 +816,19 @@ class ReplicaManager(val config: KafkaConfig,
val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
- entriesPerPartition.foreach { case (topicPartition, records) =>
+ entriesPerPartition.forKeyValue { (topicPartition, records) =>
// Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
transactionalBatches.foreach(batch => transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
- if (!transactionalBatches.isEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
+ if (transactionalBatches.nonEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
}
if (transactionalProducerInfo.size > 1) {
throw new InvalidPidMappingException("Transactional records contained more than one producer ID")
}
- def postVerificationCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
+ def postVerificationCallback(newRequestLocal: RequestLocal,
+ results: (Map[TopicPartition, Errors], Map[TopicPartition, VerificationGuard])): Unit = {
Review Comment:
The function that wraps the callback only support wrapping methods with one argument. I could not find a better way to handle this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1470098563
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -982,24 +996,21 @@ class ReplicaManager(val config: KafkaConfig,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: ((Errors, VerificationGuard)) => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
+ def generalizedCallback(results: (Map[TopicPartition, Errors], Map[TopicPartition, VerificationGuard])): Unit = {
+ val (preAppendErrors, verificationGuards) = results
+ callback((
Review Comment:
nit: are the double `((` needed? I see it elsewhere so maybe I'm missing an aspect of the language.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1448892815
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1090,38 +1090,29 @@ class ReplicaManager(val config: KafkaConfig,
* @param producerId the producer id for the producer writing to the transaction
* @param producerEpoch the epoch of the producer writing to the transaction
* @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
- * thread calling this method
* @param callback the method to execute once the verification is either completed or returns an error
*
* When the verification returns, the callback will be supplied the error if it exists or Errors.NONE.
* If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned.
- * This guard can not be used for verification and any appends that attenpt to use it will fail.
+ * This guard can not be used for verification and any appends that attempt to use it will fail.
*/
def maybeStartTransactionVerificationForPartition(
topicPartition: TopicPartition,
transactionalId: String,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: Either[Errors, VerificationGuard] => Unit
Review Comment:
I had to change the callback because `KafkaRequestHandler.wrapAsyncCallback` only support wrapping unary functions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #15176:
URL: https://github.com/apache/kafka/pull/15176#issuecomment-1914678256
@jolshan I reworked the PR based on https://github.com/apache/kafka/pull/15087. It is quite different from the previous one. Please take a look when you get a chance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1470096146
##########
core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala:
##########
@@ -935,8 +935,12 @@ private[group] class GroupCoordinator(
producerId,
producerEpoch,
RecordBatch.NO_SEQUENCE,
- requestLocal,
- postVerificationCallback
+ // Wrap the callback to be handled on an arbitrary request handler
+ // thread when transaction verification is complete.
Review Comment:
Should we leave a comment that the requestLocal passed in is only for the case where we execute immediately?
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -982,24 +996,21 @@ class ReplicaManager(val config: KafkaConfig,
producerId: Long,
producerEpoch: Short,
baseSequence: Int,
- requestLocal: RequestLocal,
- callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ callback: ((Errors, VerificationGuard)) => Unit
): Unit = {
- def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
- callback(
+ def generalizedCallback(results: (Map[TopicPartition, Errors], Map[TopicPartition, VerificationGuard])): Unit = {
+ val (preAppendErrors, verificationGuards) = results
+ callback((
Review Comment:
nit: are the double `((` needed? I see it elsewhere so maybe I'm missing an aspect of the language.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -813,19 +816,19 @@ class ReplicaManager(val config: KafkaConfig,
val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
- entriesPerPartition.foreach { case (topicPartition, records) =>
+ entriesPerPartition.forKeyValue { (topicPartition, records) =>
// Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
transactionalBatches.foreach(batch => transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
- if (!transactionalBatches.isEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
+ if (transactionalBatches.nonEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
}
if (transactionalProducerInfo.size > 1) {
throw new InvalidPidMappingException("Transactional records contained more than one producer ID")
}
- def postVerificationCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
+ def postVerificationCallback(newRequestLocal: RequestLocal,
+ results: (Map[TopicPartition, Errors], Map[TopicPartition, VerificationGuard])): Unit = {
Review Comment:
I don't have a problem with this refactor, but just curious what makes a tuple argument better than individual ones.
##########
core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorConcurrencyTest.scala:
##########
@@ -85,6 +85,10 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest
groupCoordinator = GroupCoordinator(config, replicaManager, heartbeatPurgatory, rebalancePurgatory, timer.time, metrics)
groupCoordinator.startup(() => zkClient.getTopicPartitionCount(Topic.GROUP_METADATA_TOPIC_NAME).getOrElse(config.offsetsTopicPartitions),
false)
+
+ // Transactional appends attempt to schedule to the request handler thread using
+ // a non request handler thread. Set this to avoid error.
+ KafkaRequestHandler.setBypassThreadCheck(true)
Review Comment:
Is this needed because we do the wrapper outside replica manager now?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1470109481
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -813,19 +816,19 @@ class ReplicaManager(val config: KafkaConfig,
val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
- entriesPerPartition.foreach { case (topicPartition, records) =>
+ entriesPerPartition.forKeyValue { (topicPartition, records) =>
// Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
transactionalBatches.foreach(batch => transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
- if (!transactionalBatches.isEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
+ if (transactionalBatches.nonEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
}
if (transactionalProducerInfo.size > 1) {
throw new InvalidPidMappingException("Transactional records contained more than one producer ID")
}
- def postVerificationCallback(preAppendErrors: Map[TopicPartition, Errors],
- newRequestLocal: RequestLocal,
- verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
+ def postVerificationCallback(newRequestLocal: RequestLocal,
+ results: (Map[TopicPartition, Errors], Map[TopicPartition, VerificationGuard])): Unit = {
Review Comment:
The function that wraps the callback only support wrapping methods with one argument.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14505; [6/N] Avoid recheduling callback in request thread [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15176:
URL: https://github.com/apache/kafka/pull/15176#discussion_r1454773095
##########
core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala:
##########
@@ -935,8 +935,9 @@ private[group] class GroupCoordinator(
producerId,
producerEpoch,
RecordBatch.NO_SEQUENCE,
- requestLocal,
- postVerificationCallback
+ // Wrap the callback to be handled on an arbitrary request handler thread
+ // when transaction verification is complete.
+ KafkaRequestHandler.wrapAsyncCallback(postVerificationCallback, requestLocal)
Review Comment:
We still need this in the old coordinator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org