lucasbru commented on code in PR #20663:
URL: https://github.com/apache/kafka/pull/20663#discussion_r2416157438
##########
core/src/main/scala/kafka/server/AutoTopicCreationManager.scala:
##########
@@ -173,8 +173,25 @@ class DefaultAutoTopicCreationManager(
requestContext: RequestContext,
timeoutMs: Long
): Unit = {
- if (topics.nonEmpty) {
- sendCreateTopicRequestWithErrorCaching(topics, Some(requestContext),
timeoutMs)
+ if (topics.isEmpty) {
+ return
+ }
+
+ val currentTimeMs = time.milliseconds()
+
+ // Filter out topics that are in back-off (have cached errors)
+ val cachedErrors =
topicCreationErrorCache.getErrorsForTopics(topics.keySet, currentTimeMs)
Review Comment:
That will create an extra collection and will be called fairly often. Maybe
it would be better to implement a
`topicCreationErrorCache.containsErrorsForTopic` and use this directly in the
filter, to avoid the extra collection.
##########
core/src/main/scala/kafka/server/AutoTopicCreationManager.scala:
##########
@@ -173,8 +173,25 @@ class DefaultAutoTopicCreationManager(
requestContext: RequestContext,
timeoutMs: Long
): Unit = {
- if (topics.nonEmpty) {
- sendCreateTopicRequestWithErrorCaching(topics, Some(requestContext),
timeoutMs)
+ if (topics.isEmpty) {
+ return
+ }
+
+ val currentTimeMs = time.milliseconds()
+
+ // Filter out topics that are in back-off (have cached errors)
+ val cachedErrors =
topicCreationErrorCache.getErrorsForTopics(topics.keySet, currentTimeMs)
+
+ // Filter out topics that are:
+ // 1. Already in error cache (back-off period)
+ // 2. Already in-flight (concurrent request)
+ val topicsToCreate = topics.filter { case (topicName, _) =>
+ !cachedErrors.contains(topicName) && // Not in back-off
+ inflightTopics.add(topicName) // Not in-flight
Review Comment:
Did we not add to inflightTopics before? Seems like it.
I think this is correct, but the inline comment is misaligned (I would just
remove the inline comment). If you want to keep the inline comment above, I
would add that you are actually adding this to inflight topics here.
--
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]
