RaidenE1 commented on code in PR #20325:
URL: https://github.com/apache/kafka/pull/20325#discussion_r2347933926
##########
core/src/main/scala/kafka/server/AutoTopicCreationManager.scala:
##########
@@ -50,21 +52,111 @@ 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 insertion-order removal (keeps the most recently
inserted entries)
+ */
+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] {
+ override def compare(a: Entry, b: Entry): Int =
java.lang.Long.compare(a.expirationTimeMs, b.expirationTimeMs)
+ })
+ private val lock = new ReentrantLock()
+
+ def put(topicName: String, errorMessage: String, ttlMs: Long): Unit = {
+ lock.lock()
+ try {
+ val existing = byTopic.get(topicName)
+ if (existing != null) {
+ // Remove old instance from structures
+ expiryQueue.remove(existing)
+ }
+
+ val currentTimeMs = time.milliseconds()
+ val expirationTimeMs = currentTimeMs + ttlMs
+ val entry = Entry(topicName, errorMessage, expirationTimeMs)
+ byTopic.put(topicName, entry)
+ expiryQueue.add(entry)
+
+ // Clean up expired entries
+ while (!expiryQueue.isEmpty && expiryQueue.peek().expirationTimeMs <=
currentTimeMs) {
+ val expired = expiryQueue.poll()
+ val current = byTopic.get(expired.topicName)
+ if (current != null && (current eq expired)) {
Review Comment:
I think not, it just checks if same object instance, which is O(1)
--
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
