KAFKA-19131: Adjust remote storage reader thread maximum pool size to avoid illegal argument (#19532)

The remote storage reader thread pool use same count for both maximum
and core size. If users adjust the pool size larger than original value,
it throws `IllegalArgumentException`. Updated both value to fix the
issue.

---------

Signed-off-by: PoAn Yang <[email protected]>

Reviewers: Kamal Chandraprakash <[email protected]>

Author: FrankYang0529

storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java

@@ -292,8 +292,15 @@ public void resizeExpirationThreadPool(int newSize) {
 
     public void resizeReaderThreadPool(int newSize) {
         int currentSize = remoteStorageReaderThreadPool.getCorePoolSize();
+        int currentMaximumSize = remoteStorageReaderThreadPool.getMaximumPoolSize();
         LOGGER.info("Updating remote reader thread pool size from {} to {}", currentSize, newSize);
-        remoteStorageReaderThreadPool.setCorePoolSize(newSize);
+        if (newSize > currentMaximumSize) {
+            remoteStorageReaderThreadPool.setMaximumPoolSize(newSize);
+            remoteStorageReaderThreadPool.setCorePoolSize(newSize);
+        } else {
+            remoteStorageReaderThreadPool.setCorePoolSize(newSize);
+            remoteStorageReaderThreadPool.setMaximumPoolSize(newSize);
+        }
     }
 
     private void removeMetrics() {
@@ -302,6 +309,11 @@ private void removeMetrics() {
         remoteStorageReaderThreadPool.removeMetrics();
     }
 
+    // Visible for testing
+    int readerThreadPoolSize() {
+        return remoteStorageReaderThreadPool.getCorePoolSize();
+    }
+
     /**
      * Returns the timeout for the RLM Tasks to wait for the quota to be available
      */

storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java

@@ -3753,6 +3753,15 @@ public void testMonitorableRemoteLogStorageManager() throws IOException {
         }
     }
 
+    @Test
+    void testUpdateRemoteStorageReaderThreads() {
+        assertEquals(10, remoteLogManager.readerThreadPoolSize());
+        remoteLogManager.resizeReaderThreadPool(6);
+        assertEquals(6, remoteLogManager.readerThreadPoolSize());
+        remoteLogManager.resizeReaderThreadPool(12);
+        assertEquals(12, remoteLogManager.readerThreadPoolSize());
+    }
+
     private void appendRecordsToFile(File file, int nRecords, int nRecordsPerBatch) throws IOException {
         byte magic = RecordBatch.CURRENT_MAGIC_VALUE;
         Compression compression = Compression.NONE;