junrao commented on a change in pull request #10733:
URL: https://github.com/apache/kafka/pull/10733#discussion_r652137135



##########
File path: 
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
##########
@@ -0,0 +1,350 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.log.remote.storage;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
+import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
+import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
+import static org.apache.kafka.common.config.ConfigDef.Range.between;
+import static org.apache.kafka.common.config.ConfigDef.Type.BOOLEAN;
+import static org.apache.kafka.common.config.ConfigDef.Type.DOUBLE;
+import static org.apache.kafka.common.config.ConfigDef.Type.INT;
+import static org.apache.kafka.common.config.ConfigDef.Type.LONG;
+import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
+
+public final class RemoteLogManagerConfig {
+
+    /**
+     * Prefix used for properties to be passed to {@link RemoteStorageManager} 
implementation. Remote log subsystem collects all the properties having
+     * this prefix and passes to {@code RemoteStorageManager} using {@link 
RemoteStorageManager#configure(Map)}.
+     */
+    public static final String REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP = 
"remote.log.storage.manager.impl.prefix";
+    public static final String REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_DOC = 
"Prefix used for properties to be passed to RemoteStorageManager " +
+            "implementation. For example this value can be `rsm.s3.`.";
+
+    /**
+     * Prefix used for properties to be passed to {@link 
RemoteLogMetadataManager} implementation. Remote log subsystem collects all the 
properties having
+     * this prefix and passed to {@code RemoteLogMetadataManager} using {@link 
RemoteLogMetadataManager#configure(Map)}.
+     */
+    public static final String REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP 
= "remote.log.metadata.manager.impl.prefix";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_DOC = 
"Prefix used for properties to be passed to RemoteLogMetadataManager " +
+            "implementation. For example this value can be `rlmm.s3.`.";
+
+    public static final String REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP = 
"remote.log.storage.system.enable";
+    public static final String REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC = "Whether 
to enable tier storage functionality in a broker or not. Valid values " +
+            "are `true` or `false` and the default value is false. When it is 
true broker starts all the services required for tiered storage functionality.";
+    public static final boolean DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE = 
false;
+
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP = 
"remote.log.storage.manager.class.name";
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_NAME_DOC = "Fully 
qualified class name of `RemoteLogStorageManager` implementation.";
+
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP = 
"remote.log.storage.manager.class.path";
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_PATH_DOC = "Class 
path of the `RemoteLogStorageManager` implementation." +
+            "If specified, the RemoteLogStorageManager implementation and its 
dependent libraries will be loaded by a dedicated" +
+            "classloader which searches this class path before the Kafka 
broker class path. The syntax of this parameter is same" +
+            "with the standard Java class path string.";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP = 
"remote.log.metadata.manager.class.name";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_DOC = 
"Fully qualified class name of `RemoteLogMetadataManager` implementation.";
+    //todo add the default topic based RLMM class name.
+    public static final String DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME 
= "";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP = 
"remote.log.metadata.manager.class.path";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_DOC = 
"Class path of the `RemoteLogMetadataManager` implementation." +
+            "If specified, the RemoteLogMetadataManager implementation and its 
dependent libraries will be loaded by a dedicated" +
+            "classloader which searches this class path before the Kafka 
broker class path. The syntax of this parameter is same" +
+            "with the standard Java class path string.";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP 
= "remote.log.metadata.manager.listener.name";
+    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_DOC = 
"Listener name of the local broker to which it should get connected if " +
+            "needed by RemoteLogMetadataManager implementation.";
+
+    public static final String 
REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP = 
"remote.log.index.file.cache.total.size.bytes";
+    public static final String 
REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_DOC = "The total size of the space 
allocated to store index files fetched " +
+            "from remote storage in the local storage.";
+    public static final long 
DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES = 1024 * 1024 * 1024L;
+
+    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP = 
"remote.log.manager.thread.pool.size";
+    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC = "Size 
of the thread pool used in scheduling tasks to copy " +
+            "segments, fetch remote log indexes and clean up remote log 
segments.";
+    public static final int DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE = 10;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP = 
"remote.log.manager.task.interval.ms";
+    public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_DOC = 
"Interval at which remote log manager runs the scheduled tasks like copy " +
+            "segments, and clean up remote log segments.";
+    public static final long DEFAULT_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS = 30 
* 1000L;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP 
= "remote.log.manager.task.retry.backoff.ms";
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_DOC = 
"The initial amount of wait in milli seconds before the request is retried 
again.";
+    public static final long DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS 
= 500L;
+
+    public static final String 
REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP = 
"remote.log.manager.task.retry.backoff.max.ms";
+    public static final String 
REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_DOC = "The maximum amount of time 
in milliseconds to wait when the request " +
+            "is retried again. The retry duration will increase exponentially 
for each request failure up to this maximum wait interval.";
+    public static final long 
DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS = 30 * 1000L;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP = 
"remote.log.manager.task.retry.jitter";
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_DOC = 
"Random jitter amount applied to the `remote.log.manager.task.retry.backoff.ms` 
" +
+            "for computing the resultant backoff interval. This will avoid 
thundering herds of requests.";
+    public static final double DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_JITTER = 
0.2;
+
+    public static final String REMOTE_LOG_READER_THREADS_PROP = 
"remote.log.reader.threads";
+    public static final String REMOTE_LOG_READER_THREADS_DOC = "Size of the 
thread pool that is allocated for handling remote log reads.";
+    public static final int DEFAULT_REMOTE_LOG_READER_THREADS = 10;
+
+    public static final String REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP = 
"remote.log.reader.max.pending.tasks";
+    public static final String REMOTE_LOG_READER_MAX_PENDING_TASKS_DOC = 
"Maximum remote log reader thread pool task queue size. If the task queue " +
+            "is full, fetch requests are served with an error.";
+    public static final int DEFAULT_REMOTE_LOG_READER_MAX_PENDING_TASKS = 100;
+
+    public static final ConfigDef CONFIG_DEF = new ConfigDef();
+
+    static {
+        CONFIG_DEF.define(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, BOOLEAN, 
DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE, MEDIUM, 
REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC)
+                  .define(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP, STRING, 
null, new ConfigDef.NonEmptyString(), MEDIUM, 
REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_DOC)
+                  .define(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP, 
STRING, null, new ConfigDef.NonEmptyString(),
+                          MEDIUM, 
REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_DOC)
+                  .define(REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, STRING, 
null, MEDIUM,
+                          REMOTE_STORAGE_MANAGER_CLASS_NAME_DOC)
+                  .define(REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP, STRING, 
null, MEDIUM,
+                          REMOTE_STORAGE_MANAGER_CLASS_PATH_DOC)
+                  .define(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP, STRING, 
null, MEDIUM,
+                          REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_DOC)
+                  .define(REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP, STRING, 
null, MEDIUM,
+                          REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_DOC)
+                  .define(REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP, 
STRING, null, MEDIUM,
+                          REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_DOC)
+                  .define(REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP, 
LONG,
+                          
DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES, atLeast(1), LOW,
+                          REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_DOC)
+                  .define(REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP, INT,
+                          DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE, 
atLeast(1), MEDIUM,
+                          REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC)
+                  .define(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP, LONG,
+                          DEFAULT_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS, 
atLeast(1), LOW,
+                          REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_DOC)
+                  .define(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP, LONG,
+                          DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS, 
atLeast(1), LOW,
+                          REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_DOC)
+                  .define(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP, 
LONG,
+                          
DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS, atLeast(1), LOW,
+                          REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_DOC)
+                  .define(REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP, DOUBLE,
+                          DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_JITTER, 
between(0, 0.5), LOW,
+                          REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_DOC)
+                  .define(REMOTE_LOG_READER_THREADS_PROP, INT, 
DEFAULT_REMOTE_LOG_READER_THREADS,
+                          atLeast(1), MEDIUM, REMOTE_LOG_READER_THREADS_DOC)
+                  .define(REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP, INT,
+                          DEFAULT_REMOTE_LOG_READER_MAX_PENDING_TASKS, 
atLeast(1), MEDIUM,
+                          REMOTE_LOG_READER_MAX_PENDING_TASKS_DOC);
+    }
+
+    private final boolean enableRemoteStorageSystem;
+    private final String remoteStorageManagerClassName;
+    private final String remoteStorageManagerClassPath;
+    private final String remoteLogMetadataManagerClassName;
+    private final String remoteLogMetadataManagerClassPath;
+    private final long remoteLogIndexFileCacheTotalSizeBytes;
+    private final int remoteLogManagerThreadPoolSize;
+    private final long remoteLogManagerTaskIntervalMs;
+    private final long remoteLogManagerTaskRetryBackoffMs;
+    private final long remoteLogManagerTaskRetryBackoffMaxMs;
+    private final double remoteLogManagerTaskRetryJitterMs;

Review comment:
       Jitter is now a relative thing, not ms. So we need to name the variable 
appropriately.

##########
File path: 
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
##########
@@ -0,0 +1,350 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.log.remote.storage;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
+import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
+import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
+import static org.apache.kafka.common.config.ConfigDef.Range.between;
+import static org.apache.kafka.common.config.ConfigDef.Type.BOOLEAN;
+import static org.apache.kafka.common.config.ConfigDef.Type.DOUBLE;
+import static org.apache.kafka.common.config.ConfigDef.Type.INT;
+import static org.apache.kafka.common.config.ConfigDef.Type.LONG;
+import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
+
+public final class RemoteLogManagerConfig {
+
+    /**
+     * Prefix used for properties to be passed to {@link RemoteStorageManager} 
implementation. Remote log subsystem collects all the properties having
+     * this prefix and passes to {@code RemoteStorageManager} using {@link 
RemoteStorageManager#configure(Map)}.
+     */
+    public static final String REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP = 
"remote.log.storage.manager.impl.prefix";
+    public static final String REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_DOC = 
"Prefix used for properties to be passed to RemoteStorageManager " +
+            "implementation. For example this value can be `rsm.s3.`.";
+
+    /**
+     * Prefix used for properties to be passed to {@link 
RemoteLogMetadataManager} implementation. Remote log subsystem collects all the 
properties having
+     * this prefix and passed to {@code RemoteLogMetadataManager} using {@link 
RemoteLogMetadataManager#configure(Map)}.
+     */
+    public static final String REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP 
= "remote.log.metadata.manager.impl.prefix";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_DOC = 
"Prefix used for properties to be passed to RemoteLogMetadataManager " +
+            "implementation. For example this value can be `rlmm.s3.`.";
+
+    public static final String REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP = 
"remote.log.storage.system.enable";
+    public static final String REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC = "Whether 
to enable tier storage functionality in a broker or not. Valid values " +
+            "are `true` or `false` and the default value is false. When it is 
true broker starts all the services required for tiered storage functionality.";
+    public static final boolean DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE = 
false;
+
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP = 
"remote.log.storage.manager.class.name";
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_NAME_DOC = "Fully 
qualified class name of `RemoteLogStorageManager` implementation.";
+
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP = 
"remote.log.storage.manager.class.path";
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_PATH_DOC = "Class 
path of the `RemoteLogStorageManager` implementation." +
+            "If specified, the RemoteLogStorageManager implementation and its 
dependent libraries will be loaded by a dedicated" +
+            "classloader which searches this class path before the Kafka 
broker class path. The syntax of this parameter is same" +
+            "with the standard Java class path string.";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP = 
"remote.log.metadata.manager.class.name";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_DOC = 
"Fully qualified class name of `RemoteLogMetadataManager` implementation.";
+    //todo add the default topic based RLMM class name.
+    public static final String DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME 
= "";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP = 
"remote.log.metadata.manager.class.path";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_DOC = 
"Class path of the `RemoteLogMetadataManager` implementation." +
+            "If specified, the RemoteLogMetadataManager implementation and its 
dependent libraries will be loaded by a dedicated" +
+            "classloader which searches this class path before the Kafka 
broker class path. The syntax of this parameter is same" +
+            "with the standard Java class path string.";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP 
= "remote.log.metadata.manager.listener.name";
+    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_DOC = 
"Listener name of the local broker to which it should get connected if " +
+            "needed by RemoteLogMetadataManager implementation.";
+
+    public static final String 
REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP = 
"remote.log.index.file.cache.total.size.bytes";
+    public static final String 
REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_DOC = "The total size of the space 
allocated to store index files fetched " +
+            "from remote storage in the local storage.";
+    public static final long 
DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES = 1024 * 1024 * 1024L;
+
+    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP = 
"remote.log.manager.thread.pool.size";
+    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC = "Size 
of the thread pool used in scheduling tasks to copy " +
+            "segments, fetch remote log indexes and clean up remote log 
segments.";
+    public static final int DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE = 10;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP = 
"remote.log.manager.task.interval.ms";
+    public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_DOC = 
"Interval at which remote log manager runs the scheduled tasks like copy " +
+            "segments, and clean up remote log segments.";
+    public static final long DEFAULT_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS = 30 
* 1000L;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP 
= "remote.log.manager.task.retry.backoff.ms";
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_DOC = 
"The initial amount of wait in milli seconds before the request is retried 
again.";
+    public static final long DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS 
= 500L;
+
+    public static final String 
REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP = 
"remote.log.manager.task.retry.backoff.max.ms";
+    public static final String 
REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_DOC = "The maximum amount of time 
in milliseconds to wait when the request " +
+            "is retried again. The retry duration will increase exponentially 
for each request failure up to this maximum wait interval.";
+    public static final long 
DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS = 30 * 1000L;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP = 
"remote.log.manager.task.retry.jitter";
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_DOC = 
"Random jitter amount applied to the `remote.log.manager.task.retry.backoff.ms` 
" +
+            "for computing the resultant backoff interval. This will avoid 
thundering herds of requests.";
+    public static final double DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_JITTER = 
0.2;
+
+    public static final String REMOTE_LOG_READER_THREADS_PROP = 
"remote.log.reader.threads";
+    public static final String REMOTE_LOG_READER_THREADS_DOC = "Size of the 
thread pool that is allocated for handling remote log reads.";
+    public static final int DEFAULT_REMOTE_LOG_READER_THREADS = 10;
+
+    public static final String REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP = 
"remote.log.reader.max.pending.tasks";
+    public static final String REMOTE_LOG_READER_MAX_PENDING_TASKS_DOC = 
"Maximum remote log reader thread pool task queue size. If the task queue " +
+            "is full, fetch requests are served with an error.";
+    public static final int DEFAULT_REMOTE_LOG_READER_MAX_PENDING_TASKS = 100;
+
+    public static final ConfigDef CONFIG_DEF = new ConfigDef();
+
+    static {
+        CONFIG_DEF.define(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, BOOLEAN, 
DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE, MEDIUM, 
REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC)
+                  .define(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP, STRING, 
null, new ConfigDef.NonEmptyString(), MEDIUM, 
REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_DOC)
+                  .define(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP, 
STRING, null, new ConfigDef.NonEmptyString(),

Review comment:
       Could we define those new properties as internal until the feature is 
complete?

##########
File path: 
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
##########
@@ -0,0 +1,350 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.log.remote.storage;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
+import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
+import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
+import static org.apache.kafka.common.config.ConfigDef.Range.between;
+import static org.apache.kafka.common.config.ConfigDef.Type.BOOLEAN;
+import static org.apache.kafka.common.config.ConfigDef.Type.DOUBLE;
+import static org.apache.kafka.common.config.ConfigDef.Type.INT;
+import static org.apache.kafka.common.config.ConfigDef.Type.LONG;
+import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
+
+public final class RemoteLogManagerConfig {
+
+    /**
+     * Prefix used for properties to be passed to {@link RemoteStorageManager} 
implementation. Remote log subsystem collects all the properties having
+     * this prefix and passes to {@code RemoteStorageManager} using {@link 
RemoteStorageManager#configure(Map)}.
+     */
+    public static final String REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP = 
"remote.log.storage.manager.impl.prefix";
+    public static final String REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_DOC = 
"Prefix used for properties to be passed to RemoteStorageManager " +
+            "implementation. For example this value can be `rsm.s3.`.";
+
+    /**
+     * Prefix used for properties to be passed to {@link 
RemoteLogMetadataManager} implementation. Remote log subsystem collects all the 
properties having
+     * this prefix and passed to {@code RemoteLogMetadataManager} using {@link 
RemoteLogMetadataManager#configure(Map)}.
+     */
+    public static final String REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP 
= "remote.log.metadata.manager.impl.prefix";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_DOC = 
"Prefix used for properties to be passed to RemoteLogMetadataManager " +
+            "implementation. For example this value can be `rlmm.s3.`.";
+
+    public static final String REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP = 
"remote.log.storage.system.enable";
+    public static final String REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC = "Whether 
to enable tier storage functionality in a broker or not. Valid values " +
+            "are `true` or `false` and the default value is false. When it is 
true broker starts all the services required for tiered storage functionality.";
+    public static final boolean DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE = 
false;
+
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP = 
"remote.log.storage.manager.class.name";
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_NAME_DOC = "Fully 
qualified class name of `RemoteLogStorageManager` implementation.";
+
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP = 
"remote.log.storage.manager.class.path";
+    public static final String REMOTE_STORAGE_MANAGER_CLASS_PATH_DOC = "Class 
path of the `RemoteLogStorageManager` implementation." +
+            "If specified, the RemoteLogStorageManager implementation and its 
dependent libraries will be loaded by a dedicated" +
+            "classloader which searches this class path before the Kafka 
broker class path. The syntax of this parameter is same" +
+            "with the standard Java class path string.";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP = 
"remote.log.metadata.manager.class.name";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_DOC = 
"Fully qualified class name of `RemoteLogMetadataManager` implementation.";
+    //todo add the default topic based RLMM class name.
+    public static final String DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME 
= "";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP = 
"remote.log.metadata.manager.class.path";
+    public static final String REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_DOC = 
"Class path of the `RemoteLogMetadataManager` implementation." +
+            "If specified, the RemoteLogMetadataManager implementation and its 
dependent libraries will be loaded by a dedicated" +
+            "classloader which searches this class path before the Kafka 
broker class path. The syntax of this parameter is same" +
+            "with the standard Java class path string.";
+
+    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP 
= "remote.log.metadata.manager.listener.name";
+    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_DOC = 
"Listener name of the local broker to which it should get connected if " +
+            "needed by RemoteLogMetadataManager implementation.";
+
+    public static final String 
REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP = 
"remote.log.index.file.cache.total.size.bytes";
+    public static final String 
REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_DOC = "The total size of the space 
allocated to store index files fetched " +
+            "from remote storage in the local storage.";
+    public static final long 
DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES = 1024 * 1024 * 1024L;
+
+    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP = 
"remote.log.manager.thread.pool.size";
+    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC = "Size 
of the thread pool used in scheduling tasks to copy " +
+            "segments, fetch remote log indexes and clean up remote log 
segments.";
+    public static final int DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE = 10;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP = 
"remote.log.manager.task.interval.ms";
+    public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_DOC = 
"Interval at which remote log manager runs the scheduled tasks like copy " +
+            "segments, and clean up remote log segments.";
+    public static final long DEFAULT_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS = 30 
* 1000L;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP 
= "remote.log.manager.task.retry.backoff.ms";
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_DOC = 
"The initial amount of wait in milli seconds before the request is retried 
again.";
+    public static final long DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS 
= 500L;
+
+    public static final String 
REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP = 
"remote.log.manager.task.retry.backoff.max.ms";
+    public static final String 
REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_DOC = "The maximum amount of time 
in milliseconds to wait when the request " +
+            "is retried again. The retry duration will increase exponentially 
for each request failure up to this maximum wait interval.";
+    public static final long 
DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS = 30 * 1000L;
+
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP = 
"remote.log.manager.task.retry.jitter";
+    public static final String REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_DOC = 
"Random jitter amount applied to the `remote.log.manager.task.retry.backoff.ms` 
" +

Review comment:
       Could we document that jitter is a relative thing?




-- 
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.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


Reply via email to