chia7712 commented on code in PR #20325:
URL: https://github.com/apache/kafka/pull/20325#discussion_r2379547587
##########
core/src/main/scala/kafka/server/AutoTopicCreationManager.scala:
##########
@@ -50,21 +52,96 @@ trait AutoTopicCreationManager {
def createStreamsInternalTopics(
topics: Map[String, CreatableTopic],
- requestContext: RequestContext
+ requestContext: RequestContext,
+ timeoutMs: Long
): Unit
+ def getStreamsInternalTopicCreationErrors(
+ topicNames: Set[String],
+ currentTimeMs: Long
+ ): Map[String, String]
+
+ def close(): Unit = {}
+
}
+/**
+ * Thread-safe cache that stores topic creation errors with per-entry
expiration.
+ * - Expiration: maintained by a min-heap (priority queue) on expiration time
+ * - Capacity: enforced by evicting entries with earliest expiration time (not
LRU)
+ * - Updates: old entries remain in queue but are ignored via reference
equality check
+ */
+private[server] class ExpiringErrorCache(maxSize: Int, time: Time) {
+
+ private case class Entry(topicName: String, errorMessage: String,
expirationTimeMs: Long)
+
+ private val byTopic = new ConcurrentHashMap[String, Entry]()
+ private val expiryQueue = new java.util.PriorityQueue[Entry](11, new
java.util.Comparator[Entry] {
Review Comment:
`Comparator` could be implemented by lambda.
--
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]
