jeffkbkim commented on code in PR #13267:
URL: https://github.com/apache/kafka/pull/13267#discussion_r1110474085
##########
core/src/main/scala/kafka/coordinator/transaction/ProducerIdManager.scala:
##########
@@ -123,73 +129,95 @@ class ZkProducerIdManager(brokerId: Int,
}
}
- def generateProducerId(): Long = {
+ def generateProducerId(): Try[Long] = {
this synchronized {
// grab a new block of producerIds if this block has been exhausted
if (nextProducerId > currentProducerIdBlock.lastProducerId) {
- allocateNewProducerIdBlock()
+ try {
+ allocateNewProducerIdBlock()
+ } catch {
+ case t: Throwable =>
+ return Failure(t)
+ }
nextProducerId = currentProducerIdBlock.firstProducerId
}
nextProducerId += 1
- nextProducerId - 1
+ Success(nextProducerId - 1)
+ }
+ }
+
+ override def hasValidBlock: Boolean = {
+ this synchronized {
+ !currentProducerIdBlock.equals(ProducerIdsBlock.EMPTY)
}
}
}
+/**
+ * RPCProducerIdManager allocates producer id blocks asynchronously and will
immediately fail requests
+ * for producers to retry if it does not have an available producer id and is
waiting on a new block.
+ */
class RPCProducerIdManager(brokerId: Int,
+ time: Time,
brokerEpochSupplier: () => Long,
- controllerChannel: BrokerToControllerChannelManager,
- maxWaitMs: Int) extends ProducerIdManager with
Logging {
+ controllerChannel:
BrokerToControllerChannelManager) extends ProducerIdManager with Logging {
this.logIdent = "[RPC ProducerId Manager " + brokerId + "]: "
- private val nextProducerIdBlock = new
ArrayBlockingQueue[Try[ProducerIdsBlock]](1)
+ // Visible for testing
+ private[transaction] var nextProducerIdBlock = new
AtomicReference[ProducerIdsBlock](null)
Review Comment:
i think that would work, but not sure how much more readable the code will
become. we will need to rely on the size of the queue (queueSize <= 1) to
determine whether we can fetch the next block.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
