From 4344ed625870fbd63a3846df93237bb357e47c38 Mon Sep 17 00:00:00 2001
From: Muralidhar Basani <muralidhar.basani@aiven.io>
Date: Tue, 4 Jun 2024 23:00:57 +0200
Subject: [PATCH] Refactor RemoteLogManagerConfig with AbstractConfig

---
 .../kafka/server/ReplicaManagerTest.scala     |   2 +-
 .../storage/RemoteLogManagerConfig.java       | 112 ++++++++++++++++++
 2 files changed, 113 insertions(+), 1 deletion(-)

diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index b36d3b262942f..63eb6e2dd6c58 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -32,7 +32,7 @@ import kafka.zk.KafkaZkClient
 import org.apache.kafka.clients.FetchSessionHandler
 import org.apache.kafka.common.{DirectoryId, IsolationLevel, Node, TopicIdPartition, TopicPartition, Uuid}
 import org.apache.kafka.common.compress.Compression
-import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.config.{TopicConfig}
 import org.apache.kafka.common.errors.{InvalidPidMappingException, KafkaStorageException}
 import org.apache.kafka.common.message.LeaderAndIsrRequestData
 import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
index 07bacff42271a..a81a691a2a49b 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
@@ -356,6 +356,118 @@ public final class RemoteLogManagerConfig extends AbstractConfig {
                         REMOTE_FETCH_MAX_WAIT_MS_DOC);
     }
 
+    public boolean enableRemoteStorageSystem() {
+        return getBoolean(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP);
+    }
+
+    public String remoteStorageManagerClassName() {
+        return getString(REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP);
+    }
+
+    public String remoteStorageManagerClassPath() {
+        return getString(REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP);
+    }
+
+    public String remoteLogMetadataManagerClassName() {
+        return getString(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP);
+    }
+
+    public String remoteLogMetadataManagerClassPath() {
+        return getString(REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP);
+    }
+
+    public long remoteLogIndexFileCacheTotalSizeBytes() {
+        return getLong(REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP);
+    }
+
+    public int remoteLogManagerThreadPoolSize() {
+        return getInt(REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP);
+    }
+
+    public int remoteLogManagerCopierThreadPoolSize() {
+        return getInt(REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP);
+    }
+
+    public int remoteLogManagerExpirationThreadPoolSize() {
+        return getInt(REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP);
+    }
+
+    public long remoteLogManagerTaskIntervalMs() {
+        return getLong(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP);
+    }
+
+    public long remoteLogManagerTaskRetryBackoffMs() {
+        return getLong(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP);
+    }
+
+    public long remoteLogManagerTaskRetryBackoffMaxMs() {
+        return getLong(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP);
+    }
+
+    public double remoteLogManagerTaskRetryJitter() {
+        return getDouble(REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP);
+    }
+
+    public int remoteLogReaderThreads() {
+        return getInt(REMOTE_LOG_READER_THREADS_PROP);
+    }
+
+    public int remoteLogReaderMaxPendingTasks() {
+        return getInt(REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP);
+    }
+
+    public String remoteLogMetadataManagerListenerName() {
+        return getString(REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP);
+    }
+
+    public int remoteLogMetadataCustomMetadataMaxBytes() {
+        return getInt(REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_PROP);
+    }
+
+    public String remoteStorageManagerPrefix() {
+        return getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP);
+    }
+
+    public String remoteLogMetadataManagerPrefix() {
+        return getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP);
+    }
+
+    public Map<String, Object> remoteStorageManagerProps() {
+        return Collections.unmodifiableMap(getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP) != null
+                ? originalsWithPrefix(getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP))
+                : Collections.emptyMap());
+    }
+
+    public Map<String, Object> remoteLogMetadataManagerProps() {
+        return Collections.unmodifiableMap(getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP) != null
+                ? originalsWithPrefix(getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP))
+                : Collections.emptyMap());
+    }
+
+    public long remoteLogManagerCopyMaxBytesPerSecond() {
+        return getLong(REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP);
+    }
+
+    public int remoteLogManagerCopyNumQuotaSamples() {
+        return getInt(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM_PROP);
+    }
+
+    public int remoteLogManagerCopyQuotaWindowSizeSeconds() {
+        return getInt(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS_PROP);
+    }
+
+    public long remoteLogManagerFetchMaxBytesPerSecond() {
+        return getLong(REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP);
+    }
+
+    public int remoteLogManagerFetchNumQuotaSamples() {
+        return getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_PROP);
+    }
+
+    public int remoteLogManagerFetchQuotaWindowSizeSeconds() {
+        return getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_PROP);
+    }
+
     public RemoteLogManagerConfig(Map<?, ?> props) {
         this(props, false);
     }