divijvaidya commented on code in PR #13078:
URL: https://github.com/apache/kafka/pull/13078#discussion_r1062290508
##########
core/src/main/scala/kafka/log/UnifiedLog.scala:
##########
@@ -576,6 +576,13 @@ class UnifiedLog(@volatile var logStartOffset: Long,
}
}, period = producerIdExpirationCheckIntervalMs, delay =
producerIdExpirationCheckIntervalMs, unit = TimeUnit.MILLISECONDS)
+ // Visible for testing
+ def removeExpiredProducers(currentTimeMs: Long): Unit = {
Review Comment:
could we use this method at producerExpireCheck() as well?
##########
core/src/main/scala/kafka/log/ProducerStateManager.scala:
##########
@@ -685,6 +692,7 @@ class ProducerStateManager(
if (logEndOffset != mapEndOffset) {
producers.clear()
+ _producerIdCount = 0
Review Comment:
Please move this together in a function. This coupling between the metric
value `producers` is bug prone where someone may accidentally update one
without updating the other. One way to avoid such bugs in future to always
update the producer map using methods where count would be updated too (ideally
only those functions should mutate the number of producers).
Similar comment for adding to the `producers` as well.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
