KAFKA-17928: Make remote log manager thread-pool configs dynamic

core/src/main/java/kafka/log/remote/RemoteLogManager.java

@@ -289,6 +289,21 @@ public void updateFetchQuota(long quota) {
         rlmFetchQuotaManager.updateQuota(new Quota(quota, true));
     }
 
+    public void updateCopyThreadPoolSize(int size) {
+        LOGGER.info("Updating remote copy thread pool size to {}", size);
+        rlmCopyThreadPool.resize(size);
+    }
+
+    public void updateExpirationThreadPoolSize(int size) {
+        LOGGER.info("Updating remote expiration thread pool size to {}", size);
+        rlmExpirationThreadPool.resize(size);
+    }
+
+    public void updateReaderThreadPoolSize(int size) {
+        LOGGER.info("Updating remote reader thread pool size to {}", size);
+        remoteStorageReaderThreadPool.resize(size);
+    }
+
     private void removeMetrics() {
         metricsGroup.removeMetric(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC);
         metricsGroup.removeMetric(REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC);
@@ -800,6 +815,11 @@ public void run() {
                     return;
                 }
 
+                if (!remoteLogMetadataManager.isReady(topicIdPartition)) {
+                    logger.debug("RLMM not ready for partition {}, will retry later", topicIdPartition);
+                    return;
+                }
+
                 execute(unifiedLogOptional.get());
             } catch (InterruptedException ex) {
                 if (!isCancelled()) {
@@ -2156,6 +2176,10 @@ public RLMScheduledThreadPool(int poolSize, String threadPoolName, String thread
             scheduledThreadPool = createPool();
         }
 
+        public void resize(int newSize) {
+            scheduledThreadPool.setCorePoolSize(newSize);
+        }
+
         private ScheduledThreadPoolExecutor createPool() {
             ScheduledThreadPoolExecutor threadPool = new ScheduledThreadPoolExecutor(poolSize);
             threadPool.setRemoveOnCancelPolicy(true);

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

@@ -1159,7 +1159,10 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
     newConfig.values.forEach { (k, v) =>
       if (RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP.equals(k) ||
         RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP.equals(k) ||
-        RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP.equals(k)) {
+        RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP.equals(k) ||
+        RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP.equals(k) ||
+        RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP.equals(k) ||
+        RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP.equals(k)) {
         val newValue = v.asInstanceOf[Long]
         val oldValue = getValue(server.config, k)
         if (newValue != oldValue && newValue <= 0) {
@@ -1199,14 +1202,32 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
         info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP} updated, " +
           s"old value: $oldValue, new value: $newValue")
       }
+      if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP)) {
+        val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP)
+        val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP)
+        remoteLogManager.get.updateCopyThreadPoolSize(newValue)
+      }
+      if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP)) {
+        val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP)
+        val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP)
+        remoteLogManager.get.updateExpirationThreadPoolSize(newValue)
+      }
+      if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)) {
+        val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
+        val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
+        remoteLogManager.get.updateReaderThreadPoolSize(newValue)
+      }
     }
   }
 
   private def getValue(config: KafkaConfig, name: String): Long = {
     name match {
       case RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP |
            RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP |
-           RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP =>
+           RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP |
+           RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP |
+           RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP |
+           RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP =>
         config.getLong(name)
       case n => throw new IllegalStateException(s"Unexpected dynamic remote log manager config $n")
     }
@@ -1219,6 +1240,9 @@ object DynamicRemoteLogConfig {
     RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP,
     RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP,
     RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP,
-    RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP
+    RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP,
+    RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+    RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+    RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP
   )
 }