Copilot commented on code in PR #20663:
URL: https://github.com/apache/kafka/pull/20663#discussion_r2416136294
##########
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
+ }
+
+ if (topicsToCreate.nonEmpty) {
+ sendCreateTopicRequestWithErrorCaching(topicsToCreate,
Some(requestContext), timeoutMs)
}
Review Comment:
[nitpick] Early return pattern breaks the conventional Scala style. Consider
using an if-else structure or pattern matching instead of explicit return
statements.
```suggestion
if (topics.nonEmpty) {
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
}
if (topicsToCreate.nonEmpty) {
sendCreateTopicRequestWithErrorCaching(topicsToCreate,
Some(requestContext), timeoutMs)
}
}
```
##########
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:
The filter condition has a side effect (calling `inflightTopics.add`) which
makes the code harder to reason about. Consider separating the filtering and
side-effect logic for better clarity.
```suggestion
!cachedErrors.contains(topicName) &&
!inflightTopics.contains(topicName)
}
topicsToCreate.keys.foreach(inflightTopics.add)
```
--
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]
