use linkedlist as LRU and set a default cache size

RaidenE1 · RaidenE1 · commit 5fa207ab484a · 2025-09-01T17:58:08.000-04:00
diff --git a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala
@@ -41,6 +41,7 @@ import org.apache.kafka.common.utils.Time
 import scala.collection.{Map, Seq, Set, mutable}
 import scala.jdk.CollectionConverters._
 import scala.jdk.OptionConverters.RichOptional
+import scala.util.control.Breaks._
 
 trait AutoTopicCreationManager {
 
@@ -71,6 +72,7 @@ case class CachedTopicCreationError(
   val timestamp: Long = time.milliseconds()
 }
 
+
 class DefaultAutoTopicCreationManager(
   config: KafkaConfig,
   channelManager: NodeToControllerChannelManager,
@@ -81,7 +83,20 @@ class DefaultAutoTopicCreationManager(
 ) extends AutoTopicCreationManager with Logging {
 
   private val inflightTopics = Collections.newSetFromMap(new ConcurrentHashMap[String, java.lang.Boolean]())
-  private val topicCreationErrorCache = new ConcurrentHashMap[String, CachedTopicCreationError]()
+  
+  // Use MAX_INCREMENTAL_FETCH_SESSION_CACHE_SLOTS_CONFIG as the size limit for the error cache
+  // This provides a reasonable bound (default 1000) to prevent unbounded growth
+  private val maxCacheSize = config.maxIncrementalFetchSessionCacheSlots
+  info(s"AutoTopicCreationManager initialized with error cache size limit: $maxCacheSize")
+  
+  // LRU cache with size limit to prevent unbounded memory growth
+  private val topicCreationErrorCache = Collections.synchronizedMap(
+    new java.util.LinkedHashMap[String, CachedTopicCreationError](16, 0.75f, false) {
+      override def removeEldestEntry(eldest: java.util.Map.Entry[String, CachedTopicCreationError]): Boolean = {
+        size() > maxCacheSize
+      }
+    }
+  )
 
   /**
    * Initiate auto topic creation for the given topics.
@@ -126,38 +141,59 @@ class DefaultAutoTopicCreationManager(
     }
 
     if (topics.nonEmpty) {
-      sendCreateTopicRequest(topics, Some(requestContext))
+      sendCreateTopicRequestWithErrorCaching(topics, Some(requestContext))
     }
   }
 
   override def getTopicCreationErrors(
     topicNames: Set[String],
     errorCacheTtlMs: Long
   ): Map[String, String] = {
-    val currentTime = time.milliseconds()
+    // Proactively expire old entries using the provided TTL
+    expireOldEntries(errorCacheTtlMs)
+    
     val errors = mutable.Map.empty[String, String]
-    val expiredKeys = mutable.Set.empty[String]
     
-    // Check requested topics and collect expired keys
+    // Check requested topics  
     topicNames.foreach { topicName =>
       Option(topicCreationErrorCache.get(topicName)) match {
-        case Some(cachedError) if (currentTime - cachedError.timestamp) <= errorCacheTtlMs =>
-          errors.put(topicName, cachedError.errorMessage)
-        case Some(_) =>
-          expiredKeys += topicName
+        case Some(error) =>
+          errors.put(topicName, error.errorMessage)
         case None =>
       }
     }
     
-    // Remove expired entries
-    expiredKeys.foreach { key =>
-      topicCreationErrorCache.remove(key)
-      debug(s"Removed expired topic creation error cache entry for $key")
-    }
-    
     errors.toMap
   }
 
+  /**
+   * Remove expired entries from the cache using the provided TTL.
+   * Since we use LinkedHashMap with insertion order, we only need to check 
+   * entries from the beginning until we find a non-expired entry.
+   */
+  private def expireOldEntries(ttlMs: Long): Unit = {
+    val currentTime = time.milliseconds()
+    
+    // Iterate and remove expired entries
+    val iterator = topicCreationErrorCache.entrySet().iterator()
+    
+    breakable {
+      while (iterator.hasNext) {
+        val entry = iterator.next()
+        val cachedError = entry.getValue
+        
+        if (currentTime - cachedError.timestamp > ttlMs) {
+          iterator.remove()
+          debug(s"Removed expired topic creation error cache entry for ${entry.getKey}")
+        } else {
+          // Since entries are in insertion order, if this entry is not expired,
+          // all following entries are also not expired
+          break()
+        }
+      }
+    }
+  }
+
   private def sendCreateTopicRequest(
     creatableTopics: Map[String, CreatableTopic],
     requestContext: Option[RequestContext]
@@ -182,18 +218,11 @@ class DefaultAutoTopicCreationManager(
         if (response.authenticationException() != null) {
           val authException = response.authenticationException()
           warn(s"Auto topic creation failed for ${creatableTopics.keys} with authentication exception: ${authException.getMessage}")
-          cacheTopicCreationErrors(creatableTopics.keys.toSet, authException.getMessage)
         } else if (response.versionMismatch() != null) {
           val versionException = response.versionMismatch()
           warn(s"Auto topic creation failed for ${creatableTopics.keys} with version mismatch exception: ${versionException.getMessage}")
-          cacheTopicCreationErrors(creatableTopics.keys.toSet, versionException.getMessage)
         } else {
-          response.responseBody() match {
-            case createTopicsResponse: CreateTopicsResponse =>
-              cacheTopicCreationErrorsFromResponse(createTopicsResponse)
-            case _ =>
-              debug(s"Auto topic creation completed for ${creatableTopics.keys} with response ${response.responseBody}.")
-          }
+          debug(s"Auto topic creation completed for ${creatableTopics.keys} with response ${response.responseBody}.")
         }
       }
     }
@@ -319,6 +348,80 @@ class DefaultAutoTopicCreationManager(
     (creatableTopics, uncreatableTopics)
   }
 
+  private def sendCreateTopicRequestWithErrorCaching(
+    creatableTopics: Map[String, CreatableTopic],
+    requestContext: Option[RequestContext]
+  ): Seq[MetadataResponseTopic] = {
+    val topicsToCreate = new CreateTopicsRequestData.CreatableTopicCollection(creatableTopics.size)
+    topicsToCreate.addAll(creatableTopics.values.asJavaCollection)
+
+    val createTopicsRequest = new CreateTopicsRequest.Builder(
+      new CreateTopicsRequestData()
+        .setTimeoutMs(config.requestTimeoutMs)
+        .setTopics(topicsToCreate)
+    )
+
+    val requestCompletionHandler = new ControllerRequestCompletionHandler {
+      override def onTimeout(): Unit = {
+        clearInflightRequests(creatableTopics)
+        debug(s"Auto topic creation timed out for ${creatableTopics.keys}.")
+      }
+
+      override def onComplete(response: ClientResponse): Unit = {
+        clearInflightRequests(creatableTopics)
+        if (response.authenticationException() != null) {
+          val authException = response.authenticationException()
+          warn(s"Auto topic creation failed for ${creatableTopics.keys} with authentication exception: ${authException.getMessage}")
+          cacheTopicCreationErrors(creatableTopics.keys.toSet, authException.getMessage)
+        } else if (response.versionMismatch() != null) {
+          val versionException = response.versionMismatch()
+          warn(s"Auto topic creation failed for ${creatableTopics.keys} with version mismatch exception: ${versionException.getMessage}")
+          cacheTopicCreationErrors(creatableTopics.keys.toSet, versionException.getMessage)
+        } else {
+          response.responseBody() match {
+            case createTopicsResponse: CreateTopicsResponse =>
+              cacheTopicCreationErrorsFromResponse(createTopicsResponse)
+            case _ =>
+              debug(s"Auto topic creation completed for ${creatableTopics.keys} with response ${response.responseBody}.")
+          }
+        }
+      }
+    }
+
+    val request = requestContext.map { context =>
+      val requestVersion =
+        channelManager.controllerApiVersions.toScala match {
+          case None =>
+            // We will rely on the Metadata request to be retried in the case
+            // that the latest version is not usable by the controller.
+            ApiKeys.CREATE_TOPICS.latestVersion()
+          case Some(nodeApiVersions) =>
+            nodeApiVersions.latestUsableVersion(ApiKeys.CREATE_TOPICS)
+        }
+
+      // Borrow client information such as client id and correlation id from the original request,
+      // in order to correlate the create request with the original metadata request.
+      val requestHeader = new RequestHeader(ApiKeys.CREATE_TOPICS,
+        requestVersion,
+        context.clientId,
+        context.correlationId)
+      ForwardingManager.buildEnvelopeRequest(context,
+        createTopicsRequest.build(requestVersion).serializeWithHeader(requestHeader))
+    }.getOrElse(createTopicsRequest)
+
+    channelManager.sendRequest(request, requestCompletionHandler)
+
+    val creatableTopicResponses = creatableTopics.keySet.toSeq.map { topic =>
+      new MetadataResponseTopic()
+        .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
+        .setName(topic)
+        .setIsInternal(Topic.isInternal(topic))
+    }
+
+    info(s"Sent auto-creation request for ${creatableTopics.keys} to the active controller.")
+    creatableTopicResponses
+  }
+
   private def cacheTopicCreationErrors(topicNames: Set[String], errorMessage: String): Unit = {
     topicNames.foreach { topicName =>
       topicCreationErrorCache.put(topicName, CachedTopicCreationError(errorMessage, time))
@@ -337,7 +440,6 @@ class DefaultAutoTopicCreationManager(
         )
         debug(s"Cached topic creation error for ${topicResult.name()}: $errorMessage")
       }
-      
     }
   }
 
diff --git a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala
@@ -511,12 +511,122 @@ class AutoTopicCreationManagerTest {
       shareCoordinator,
       mockTime)
 
-    val defaultTtlMs = config.groupCoordinatorConfig.streamsGroupSessionTimeoutMs()
-    val initialResult = autoTopicCreationManager.getTopicCreationErrors(Set("nonexistent-topic"), defaultTtlMs)
-    assertTrue(initialResult.isEmpty)
 
-    mockTime.sleep(config.groupCoordinatorConfig.classicGroupMaxSessionTimeoutMs + 1000)
-    val laterResult = autoTopicCreationManager.getTopicCreationErrors(Set("nonexistent-topic"), defaultTtlMs)
-    assertTrue(laterResult.isEmpty)
+    // First cache an error by simulating topic creation failure
+    val topics = Map(
+      "test-topic" -> new CreatableTopic().setName("test-topic").setNumPartitions(1).setReplicationFactor(1)
+    )
+    val requestContext = initializeRequestContextWithUserPrincipal()
+    autoTopicCreationManager.createStreamsInternalTopics(topics, requestContext)
+
+    val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
+    Mockito.verify(brokerToController).sendRequest(
+      any(classOf[AbstractRequest.Builder[_ <: AbstractRequest]]),
+      argumentCaptor.capture())
+
+    // Simulate a CreateTopicsResponse with error
+    val createTopicsResponseData = new org.apache.kafka.common.message.CreateTopicsResponseData()
+    val topicResult = new org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult()
+      .setName("test-topic")
+      .setErrorCode(Errors.INVALID_REPLICATION_FACTOR.code())
+      .setErrorMessage("Invalid replication factor")
+    createTopicsResponseData.topics().add(topicResult)
+
+    val createTopicsResponse = new CreateTopicsResponse(createTopicsResponseData)
+    val header = new RequestHeader(ApiKeys.CREATE_TOPICS, 0, "client", 1)
+    val clientResponse = new ClientResponse(header, null, null,
+      0, 0, false, null, null, createTopicsResponse)
+
+    // Cache the error at T0
+    argumentCaptor.getValue.asInstanceOf[ControllerRequestCompletionHandler].onComplete(clientResponse)
+
+    val shortTtlMs = 1000L // Use 1 second TTL for faster testing
+    
+    // Verify error is cached and accessible within TTL
+    val cachedErrors = autoTopicCreationManager.getTopicCreationErrors(Set("test-topic"), shortTtlMs)
+    assertEquals(1, cachedErrors.size)
+    assertEquals("Invalid replication factor", cachedErrors("test-topic"))
+
+    // Advance time beyond TTL
+    mockTime.sleep(shortTtlMs + 100) // T0 + 1.1 seconds
+
+    // Verify error is now expired and proactively cleaned up
+    val expiredErrors = autoTopicCreationManager.getTopicCreationErrors(Set("test-topic"), shortTtlMs)
+    assertTrue(expiredErrors.isEmpty, "Expired errors should be proactively cleaned up")
+  }
+
+  @Test
+  def testErrorCacheLRUEviction(): Unit = {
+    // Create a config with a small cache size for testing
+    val props = TestUtils.createBrokerConfig(1)
+    props.setProperty(ServerConfigs.REQUEST_TIMEOUT_MS_CONFIG, requestTimeout.toString)
+    props.setProperty(ServerConfigs.MAX_INCREMENTAL_FETCH_SESSION_CACHE_SLOTS_CONFIG, "3") // Small cache size for testing
+    
+    props.setProperty(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, internalTopicPartitions.toString)
+    props.setProperty(TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, internalTopicPartitions.toString)
+    props.setProperty(ShareCoordinatorConfig.STATE_TOPIC_REPLICATION_FACTOR_CONFIG , internalTopicPartitions.toString)
+    
+    props.setProperty(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, internalTopicReplicationFactor.toString)
+    props.setProperty(TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG, internalTopicReplicationFactor.toString)
+    props.setProperty(ShareCoordinatorConfig.STATE_TOPIC_NUM_PARTITIONS_CONFIG, internalTopicReplicationFactor.toString)
+    
+    val smallCacheConfig = KafkaConfig.fromProps(props)
+    
+    // Verify the configuration was properly set
+    assertEquals(3, smallCacheConfig.maxIncrementalFetchSessionCacheSlots, "Cache size configuration should be 3")
+    
+    // Replace the test class's config with our smallCacheConfig
+    // so that initializeRequestContext will use the correct config
+    config = smallCacheConfig
+    
+    val requestContext = initializeRequestContextWithUserPrincipal()
+    
+    // Create 5 topics to exceed the cache size of 3
+    val topicNames = (1 to 5).map(i => s"test-topic-$i")
+    
+    // Add errors for all 5 topics to the cache
+    topicNames.zipWithIndex.foreach { case (topicName, idx) =>
+      val topics = Map(
+        topicName -> new CreatableTopic().setName(topicName).setNumPartitions(1).setReplicationFactor(1)
+      )
+      
+      autoTopicCreationManager.createStreamsInternalTopics(topics, requestContext)
+      
+      val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
+      Mockito.verify(brokerToController, Mockito.atLeastOnce()).sendRequest(
+        any(classOf[AbstractRequest.Builder[_ <: AbstractRequest]]),
+        argumentCaptor.capture())
+      
+      // Simulate error response for this topic
+      val createTopicsResponseData = new org.apache.kafka.common.message.CreateTopicsResponseData()
+      val topicResult = new org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult()
+        .setName(topicName)
+        .setErrorCode(Errors.TOPIC_ALREADY_EXISTS.code())
+        .setErrorMessage(s"Topic '$topicName' already exists.")
+      createTopicsResponseData.topics().add(topicResult)
+      
+      val createTopicsResponse = new CreateTopicsResponse(createTopicsResponseData)
+      val header = new RequestHeader(ApiKeys.CREATE_TOPICS, 0, "client", 1)
+      val clientResponse = new ClientResponse(header, null, null,
+        0, 0, false, null, null, createTopicsResponse)
+      
+      argumentCaptor.getValue.asInstanceOf[ControllerRequestCompletionHandler].onComplete(clientResponse)
+      
+      // Advance time slightly between additions to ensure different timestamps
+      mockTime.sleep(10)
+      
+    }
+    
+    // With cache size of 3, topics 1 and 2 should have been evicted
+    val longTtlMs = 60000L // Use a long TTL to ensure entries aren't expired by time
+    val cachedErrors = autoTopicCreationManager.getTopicCreationErrors(topicNames.toSet, longTtlMs)
+    
+    // Only the last 3 topics should be in the cache (topics 3, 4, 5)
+    assertEquals(3, cachedErrors.size, "Cache should contain only the most recent 3 entries")
+    assertTrue(cachedErrors.contains("test-topic-3"), "test-topic-3 should be in cache")
+    assertTrue(cachedErrors.contains("test-topic-4"), "test-topic-4 should be in cache")
+    assertTrue(cachedErrors.contains("test-topic-5"), "test-topic-5 should be in cache")
+    assertTrue(!cachedErrors.contains("test-topic-1"), "test-topic-1 should have been evicted")
+    assertTrue(!cachedErrors.contains("test-topic-2"), "test-topic-2 should have been evicted")
   }
 }
