Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kamalcph merged 3 commits into from
Dec 14, 2024
Merged

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

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d0c23d8
KAFKA-17928: Make remote log manager thread-pool configs dynamic
peterxcli Nov 19, 2024
117d258
Disallow configuring -1 for copier and expiration thread pools
kamalcph Dec 12, 2024
19288ac
Addressed the review comments
kamalcph Dec 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 38 additions & 23 deletions core/src/main/java/kafka/log/remote/RemoteLogManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,8 @@
import org.apache.kafka.common.utils.BufferSupplier;
import org.apache.kafka.common.utils.ChildFirstClassLoader;
import org.apache.kafka.common.utils.CloseableIterator;
import org.apache.kafka.common.utils.KafkaThread;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.ThreadUtils;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.server.common.CheckpointFile;
Expand Down Expand Up @@ -128,9 +128,7 @@
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
Expand Down Expand Up @@ -162,7 +160,7 @@
public class RemoteLogManager implements Closeable {

private static final Logger LOGGER = LoggerFactory.getLogger(RemoteLogManager.class);
private static final String REMOTE_LOG_READER_THREAD_NAME_PREFIX = "remote-log-reader";
private static final String REMOTE_LOG_READER_THREAD_NAME_PATTERN = "remote-log-reader-%d";
private final RemoteLogManagerConfig rlmConfig;
private final int brokerId;
private final String logDir;
Expand Down Expand Up @@ -255,18 +253,18 @@ public RemoteLogManager(RemoteLogManagerConfig rlmConfig,
indexCache = new RemoteIndexCache(rlmConfig.remoteLogIndexFileCacheTotalSizeBytes(), remoteLogStorageManager, logDir);
delayInMs = rlmConfig.remoteLogManagerTaskIntervalMs();
rlmCopyThreadPool = new RLMScheduledThreadPool(rlmConfig.remoteLogManagerCopierThreadPoolSize(),
"RLMCopyThreadPool", "kafka-rlm-copy-thread-pool-");
"RLMCopyThreadPool", "kafka-rlm-copy-thread-pool-%d");
rlmExpirationThreadPool = new RLMScheduledThreadPool(rlmConfig.remoteLogManagerExpirationThreadPoolSize(),
"RLMExpirationThreadPool", "kafka-rlm-expiration-thread-pool-");
"RLMExpirationThreadPool", "kafka-rlm-expiration-thread-pool-%d");
followerThreadPool = new RLMScheduledThreadPool(rlmConfig.remoteLogManagerThreadPoolSize(),
"RLMFollowerScheduledThreadPool", "kafka-rlm-follower-thread-pool-");
"RLMFollowerScheduledThreadPool", "kafka-rlm-follower-thread-pool-%d");

metricsGroup.newGauge(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC, rlmCopyThreadPool::getIdlePercent);
remoteReadTimer = metricsGroup.newTimer(REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC,
TimeUnit.MILLISECONDS, TimeUnit.SECONDS);

remoteStorageReaderThreadPool = new RemoteStorageThreadPool(
REMOTE_LOG_READER_THREAD_NAME_PREFIX,
REMOTE_LOG_READER_THREAD_NAME_PATTERN,
rlmConfig.remoteLogReaderThreads(),
rlmConfig.remoteLogReaderMaxPendingTasks()
);
Expand All @@ -290,6 +288,24 @@ public void updateFetchQuota(long quota) {
rlmFetchQuotaManager.updateQuota(new Quota(quota, true));
}

public void resizeCopierThreadPool(int newSize) {
int currentSize = rlmCopyThreadPool.getCorePoolSize();
LOGGER.info("Updating remote copy thread pool size from {} to {}", currentSize, newSize);
rlmCopyThreadPool.setCorePoolSize(newSize);
}

public void resizeExpirationThreadPool(int newSize) {
int currentSize = rlmExpirationThreadPool.getCorePoolSize();
LOGGER.info("Updating remote expiration thread pool size from {} to {}", currentSize, newSize);
rlmExpirationThreadPool.setCorePoolSize(newSize);
}

public void resizeReaderThreadPool(int newSize) {
int currentSize = remoteStorageReaderThreadPool.getCorePoolSize();
LOGGER.info("Updating remote reader thread pool size from {} to {}", currentSize, newSize);
remoteStorageReaderThreadPool.setCorePoolSize(newSize);
}

private void removeMetrics() {
metricsGroup.removeMetric(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC);
metricsGroup.removeMetric(REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC);
Expand Down Expand Up @@ -2152,31 +2168,30 @@ RLMTaskWithFuture followerTask(TopicIdPartition partition) {
static class RLMScheduledThreadPool {

private static final Logger LOGGER = LoggerFactory.getLogger(RLMScheduledThreadPool.class);
private final int poolSize;
private final String threadPoolName;
private final String threadNamePrefix;
private final String threadNamePattern;
private final ScheduledThreadPoolExecutor scheduledThreadPool;

public RLMScheduledThreadPool(int poolSize, String threadPoolName, String threadNamePrefix) {
this.poolSize = poolSize;
public RLMScheduledThreadPool(int poolSize, String threadPoolName, String threadNamePattern) {
this.threadPoolName = threadPoolName;
this.threadNamePrefix = threadNamePrefix;
scheduledThreadPool = createPool();
this.threadNamePattern = threadNamePattern;
scheduledThreadPool = createPool(poolSize);
}

public void setCorePoolSize(int newSize) {
scheduledThreadPool.setCorePoolSize(newSize);
}

public int getCorePoolSize() {
return scheduledThreadPool.getCorePoolSize();
}

private ScheduledThreadPoolExecutor createPool() {
private ScheduledThreadPoolExecutor createPool(int poolSize) {
ScheduledThreadPoolExecutor threadPool = new ScheduledThreadPoolExecutor(poolSize);
threadPool.setRemoveOnCancelPolicy(true);
threadPool.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
threadPool.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
threadPool.setThreadFactory(new ThreadFactory() {
private final AtomicInteger sequence = new AtomicInteger();

public Thread newThread(Runnable r) {
return KafkaThread.daemon(threadNamePrefix + sequence.incrementAndGet(), r);
}
});

threadPool.setThreadFactory(ThreadUtils.createThreadFactory(threadNamePattern, true));
Comment thread
kamalcph marked this conversation as resolved.
Outdated
Comment thread
kamalcph marked this conversation as resolved.
Outdated
return threadPool;
}

Expand Down
42 changes: 36 additions & 6 deletions core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1167,6 +1167,22 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
throw new ConfigException(s"$errorMsg, value should be at least 1")
}
}

if (RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_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)) {
val newValue = v.asInstanceOf[Int]
val oldValue = server.config.getInt(k)
if (newValue != oldValue) {
val errorMsg = s"Dynamic thread count update validation failed for $k=$v"
if (newValue <= 0)
Comment thread
kamalcph marked this conversation as resolved.
throw new ConfigException(s"$errorMsg, value should be at least 1")
if (newValue < oldValue / 2)
throw new ConfigException(s"$errorMsg, value should be at least half the current value $oldValue")
if (newValue > oldValue * 2)
throw new ConfigException(s"$errorMsg, value should not be greater than double the current value $oldValue")
}
}
}
}

Expand All @@ -1176,29 +1192,40 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w

def isChangedLongValue(k : String): Boolean = oldLongValue(k) != newLongValue(k)

val remoteLogManager = server.remoteLogManagerOpt
if (remoteLogManager.nonEmpty) {
if (server.remoteLogManagerOpt.nonEmpty) {
val remoteLogManager = server.remoteLogManagerOpt.get
if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP)) {
val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP)
val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP)
remoteLogManager.get.resizeCacheSize(newValue)
remoteLogManager.resizeCacheSize(newValue)
info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP} updated, " +
s"old value: $oldValue, new value: $newValue")
}
if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP)) {
val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP)
val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP)
remoteLogManager.get.updateCopyQuota(newValue)
remoteLogManager.updateCopyQuota(newValue)
info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP} updated, " +
s"old value: $oldValue, new value: $newValue")
}
if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP)) {
val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP)
val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP)
remoteLogManager.get.updateFetchQuota(newValue)
remoteLogManager.updateFetchQuota(newValue)
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")
}

val newRLMConfig = newConfig.remoteLogManagerConfig
val oldRLMConfig = oldConfig.remoteLogManagerConfig
if (newRLMConfig.remoteLogManagerCopierThreadPoolSize() != oldRLMConfig.remoteLogManagerCopierThreadPoolSize())
remoteLogManager.resizeCopierThreadPool(newRLMConfig.remoteLogManagerCopierThreadPoolSize())

if (newRLMConfig.remoteLogManagerExpirationThreadPoolSize() != oldRLMConfig.remoteLogManagerExpirationThreadPoolSize())
remoteLogManager.resizeExpirationThreadPool(newRLMConfig.remoteLogManagerExpirationThreadPoolSize())

if (newRLMConfig.remoteLogReaderThreads() != oldRLMConfig.remoteLogReaderThreads())
remoteLogManager.resizeReaderThreadPool(newRLMConfig.remoteLogReaderThreads())
}
}

Expand All @@ -1219,6 +1246,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
)
}
75 changes: 75 additions & 0 deletions core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,6 +202,81 @@ class DynamicBrokerConfigTest {
)
}

@Test
def testUpdateRemoteLogManagerDynamicThreadPool(): Unit = {
val origProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
val config = KafkaConfig(origProps)
assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE, config.remoteLogManagerConfig.remoteLogManagerCopierThreadPoolSize())
assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE, config.remoteLogManagerConfig.remoteLogManagerExpirationThreadPoolSize())
assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_READER_THREADS, config.remoteLogManagerConfig.remoteLogReaderThreads())

val serverMock = mock(classOf[KafkaBroker])
val remoteLogManager = mock(classOf[RemoteLogManager])
when(serverMock.config).thenReturn(config)
when(serverMock.remoteLogManagerOpt).thenReturn(Some(remoteLogManager))

config.dynamicConfig.initialize(None, None)
config.dynamicConfig.addBrokerReconfigurable(new DynamicRemoteLogConfig(serverMock))

// Test dynamic update with valid values
val props = new Properties()
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP, "8")
config.dynamicConfig.validate(props, perBrokerConfig = true)
config.dynamicConfig.updateDefaultConfig(props)
assertEquals(8, config.remoteLogManagerConfig.remoteLogManagerCopierThreadPoolSize())
verify(remoteLogManager).resizeCopierThreadPool(8)

props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP, "7")
config.dynamicConfig.validate(props, perBrokerConfig = false)
config.dynamicConfig.updateDefaultConfig(props)
assertEquals(7, config.remoteLogManagerConfig.remoteLogManagerExpirationThreadPoolSize())
verify(remoteLogManager).resizeExpirationThreadPool(7)

props.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, "6")
config.dynamicConfig.validate(props, perBrokerConfig = true)
config.dynamicConfig.updateDefaultConfig(props)
assertEquals(6, config.remoteLogManagerConfig.remoteLogReaderThreads())
verify(remoteLogManager).resizeReaderThreadPool(6)
props.clear()
verifyNoMoreInteractions(remoteLogManager)
}

@Test
def testRemoteLogDynamicThreadPoolWithInvalidValues(): Unit = {
val origProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
val config = KafkaConfig(origProps)

val serverMock = mock(classOf[KafkaBroker])
val remoteLogManager = mock(classOf[RemoteLogManager])
when(serverMock.config).thenReturn(config)
when(serverMock.remoteLogManagerOpt).thenReturn(Some(remoteLogManager))

config.dynamicConfig.initialize(None, None)
config.dynamicConfig.addBrokerReconfigurable(new DynamicRemoteLogConfig(serverMock))

// Test dynamic update with invalid values
val props = new Properties()
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP, "0")
val err = assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props, perBrokerConfig = true))
assertTrue(err.getMessage.contains("Value must be at least 1"))

val props1 = new Properties()
props1.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP, "-1")
val err1 = assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props1, perBrokerConfig = false))
assertTrue(err1.getMessage.contains("Value must be at least 1"))

val props2 = new Properties()
props2.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, "2")
val err2 = assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props2, perBrokerConfig = false))
assertTrue(err2.getMessage.contains("value should be at least half the current value"))

val props3 = new Properties()
props3.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, "-1")
val err3 = assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props, perBrokerConfig = true))

@chia7712 chia7712 May 23, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It should be props3 rather than props. I noticed this when reviewing #22349 :)

we will file a small patch for it

@FrankYang0529 FrankYang0529 May 23, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#22352

assertTrue(err3.getMessage.contains("Value must be at least 1"))
verifyNoMoreInteractions(remoteLogManager)
}

@nowarn("cat=deprecation")
@Test
def testConfigUpdateWithSomeInvalidConfigs(): Unit = {
Expand Down
4 changes: 2 additions & 2 deletions core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1102,8 +1102,8 @@ class KafkaConfigTest {
case RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP => // ignore string
case RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -2)
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -2)
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1, -2)
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1, -2)
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
Expand Down
11 changes: 4 additions & 7 deletions storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,21 +93,18 @@ public final class RemoteLogManagerConfig {
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 = "Deprecated. 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 String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling follower tasks to read " +
"the highest-uploaded remote-offset for follower partitions.";
public static final int DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE = 2;

private static final String REMOTE_LOG_MANAGER_THREAD_POOL_FALLBACK = "The default value of -1 means that this will be set to the configured value of " +
REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP + ", if available; otherwise, it defaults to " + DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE + ".";

public static final String REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP = "remote.log.manager.copier.thread.pool.size";
public static final String REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling tasks " +
"to copy segments. " + REMOTE_LOG_MANAGER_THREAD_POOL_FALLBACK;
"to copy segments.";
public static final int DEFAULT_REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE = 10;

public static final String REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP = "remote.log.manager.expiration.thread.pool.size";
public static final String REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling tasks " +
"to clean up remote log segments. " + REMOTE_LOG_MANAGER_THREAD_POOL_FALLBACK;
"to clean up the expired remote log segments.";
public static final int DEFAULT_REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE = 10;

public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP = "remote.log.manager.task.interval.ms";
Expand Down
Loading