Skip to content

Commit 587f440

Browse files
peterxclikamalcph
peterxcli
and
kamalcph
committed
KAFKA-17928: Make remote log manager thread-pool configs dynamic
Co-authored-by: Kamal Chandraprakash <kamal.chandraprakash@gmail.com>
1 parent 441a6d0 commit 587f440

4 files changed

Lines changed: 122 additions & 43 deletions

File tree

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

Lines changed: 34 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,8 @@
4242
import org.apache.kafka.common.utils.BufferSupplier;
4343
import org.apache.kafka.common.utils.ChildFirstClassLoader;
4444
import org.apache.kafka.common.utils.CloseableIterator;
45-
import org.apache.kafka.common.utils.KafkaThread;
4645
import org.apache.kafka.common.utils.LogContext;
46+
import org.apache.kafka.common.utils.ThreadUtils;
4747
import org.apache.kafka.common.utils.Time;
4848
import org.apache.kafka.common.utils.Utils;
4949
import org.apache.kafka.server.common.CheckpointFile;
@@ -127,9 +127,7 @@
127127
import java.util.concurrent.Future;
128128
import java.util.concurrent.ScheduledFuture;
129129
import java.util.concurrent.ScheduledThreadPoolExecutor;
130-
import java.util.concurrent.ThreadFactory;
131130
import java.util.concurrent.TimeUnit;
132-
import java.util.concurrent.atomic.AtomicInteger;
133131
import java.util.concurrent.locks.Condition;
134132
import java.util.concurrent.locks.ReentrantLock;
135133
import java.util.function.BiConsumer;
@@ -161,7 +159,7 @@
161159
public class RemoteLogManager implements Closeable {
162160

163161
private static final Logger LOGGER = LoggerFactory.getLogger(RemoteLogManager.class);
164-
private static final String REMOTE_LOG_READER_THREAD_NAME_PREFIX = "remote-log-reader";
162+
private static final String REMOTE_LOG_READER_THREAD_NAME_PREFIX = "remote-log-reader-%d";
165163
private final RemoteLogManagerConfig rlmConfig;
166164
private final int brokerId;
167165
private final String logDir;
@@ -254,11 +252,11 @@ public RemoteLogManager(RemoteLogManagerConfig rlmConfig,
254252
indexCache = new RemoteIndexCache(rlmConfig.remoteLogIndexFileCacheTotalSizeBytes(), remoteLogStorageManager, logDir);
255253
delayInMs = rlmConfig.remoteLogManagerTaskIntervalMs();
256254
rlmCopyThreadPool = new RLMScheduledThreadPool(rlmConfig.remoteLogManagerCopierThreadPoolSize(),
257-
"RLMCopyThreadPool", "kafka-rlm-copy-thread-pool-");
255+
"RLMCopyThreadPool", "kafka-rlm-copy-thread-pool-%d");
258256
rlmExpirationThreadPool = new RLMScheduledThreadPool(rlmConfig.remoteLogManagerExpirationThreadPoolSize(),
259-
"RLMExpirationThreadPool", "kafka-rlm-expiration-thread-pool-");
257+
"RLMExpirationThreadPool", "kafka-rlm-expiration-thread-pool-%d");
260258
followerThreadPool = new RLMScheduledThreadPool(rlmConfig.remoteLogManagerThreadPoolSize(),
261-
"RLMFollowerScheduledThreadPool", "kafka-rlm-follower-thread-pool-");
259+
"RLMFollowerScheduledThreadPool", "kafka-rlm-follower-thread-pool-%d");
262260

263261
metricsGroup.newGauge(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC, rlmCopyThreadPool::getIdlePercent);
264262
remoteReadTimer = metricsGroup.newTimer(REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC,
@@ -289,6 +287,24 @@ public void updateFetchQuota(long quota) {
289287
rlmFetchQuotaManager.updateQuota(new Quota(quota, true));
290288
}
291289

290+
public void resizeCopierThreadPool(int newSize) {
291+
int currentSize = rlmCopyThreadPool.getCorePoolSize();
292+
LOGGER.info("Updating remote copy thread pool size from {} to {}", currentSize, newSize);
293+
rlmCopyThreadPool.setCorePoolSize(newSize);
294+
}
295+
296+
public void resizeExpirationThreadPool(int newSize) {
297+
int currentSize = rlmExpirationThreadPool.getCorePoolSize();
298+
LOGGER.info("Updating remote expiration thread pool size from {} to {}", currentSize, newSize);
299+
rlmExpirationThreadPool.setCorePoolSize(newSize);
300+
}
301+
302+
public void resizeReaderThreadPool(int newSize) {
303+
int currentSize = remoteStorageReaderThreadPool.getCorePoolSize();
304+
LOGGER.info("Updating remote reader thread pool size from {} to {}", currentSize, newSize);
305+
remoteStorageReaderThreadPool.setCorePoolSize(newSize);
306+
}
307+
292308
private void removeMetrics() {
293309
metricsGroup.removeMetric(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC);
294310
metricsGroup.removeMetric(REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC);
@@ -2151,31 +2167,30 @@ RLMTaskWithFuture followerTask(TopicIdPartition partition) {
21512167
static class RLMScheduledThreadPool {
21522168

21532169
private static final Logger LOGGER = LoggerFactory.getLogger(RLMScheduledThreadPool.class);
2154-
private final int poolSize;
21552170
private final String threadPoolName;
21562171
private final String threadNamePrefix;
21572172
private final ScheduledThreadPoolExecutor scheduledThreadPool;
21582173

21592174
public RLMScheduledThreadPool(int poolSize, String threadPoolName, String threadNamePrefix) {
2160-
this.poolSize = poolSize;
21612175
this.threadPoolName = threadPoolName;
21622176
this.threadNamePrefix = threadNamePrefix;
2163-
scheduledThreadPool = createPool();
2177+
scheduledThreadPool = createPool(poolSize);
2178+
}
2179+
2180+
public void setCorePoolSize(int newSize) {
2181+
scheduledThreadPool.setCorePoolSize(newSize);
2182+
}
2183+
2184+
public int getCorePoolSize() {
2185+
return scheduledThreadPool.getCorePoolSize();
21642186
}
21652187

2166-
private ScheduledThreadPoolExecutor createPool() {
2188+
private ScheduledThreadPoolExecutor createPool(int poolSize) {
21672189
ScheduledThreadPoolExecutor threadPool = new ScheduledThreadPoolExecutor(poolSize);
21682190
threadPool.setRemoveOnCancelPolicy(true);
21692191
threadPool.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
21702192
threadPool.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
2171-
threadPool.setThreadFactory(new ThreadFactory() {
2172-
private final AtomicInteger sequence = new AtomicInteger();
2173-
2174-
public Thread newThread(Runnable r) {
2175-
return KafkaThread.daemon(threadNamePrefix + sequence.incrementAndGet(), r);
2176-
}
2177-
});
2178-
2193+
threadPool.setThreadFactory(ThreadUtils.createThreadFactory(threadNamePrefix, true));
21792194
return threadPool;
21802195
}
21812196

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

Lines changed: 37 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1167,6 +1167,22 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
11671167
throw new ConfigException(s"$errorMsg, value should be at least 1")
11681168
}
11691169
}
1170+
1171+
if (RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP.equals(k) ||
1172+
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP.equals(k) ||
1173+
RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP.equals(k)) {
1174+
val newValue = v.asInstanceOf[Int]
1175+
val oldValue = server.config.getInt(k)
1176+
if (newValue != oldValue) {
1177+
val errorMsg = s"Dynamic thread count update validation failed for $k=$v"
1178+
if (newValue <= 0)
1179+
throw new ConfigException(s"$errorMsg, value should be at least 1")
1180+
if (oldValue != -1 && newValue < oldValue / 2)
1181+
throw new ConfigException(s"$errorMsg, value should be at least half the current value $oldValue")
1182+
if (oldValue != -1 && newValue > oldValue * 2)
1183+
throw new ConfigException(s"$errorMsg, value should not be greater than double the current value $oldValue")
1184+
}
1185+
}
11701186
}
11711187
}
11721188

@@ -1176,29 +1192,41 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
11761192

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

1179-
val remoteLogManager = server.remoteLogManagerOpt
1180-
if (remoteLogManager.nonEmpty) {
1195+
val remoteLogManagerOpt = server.remoteLogManagerOpt
1196+
if (remoteLogManagerOpt.nonEmpty) {
1197+
val remoteLogManager = remoteLogManagerOpt.get
11811198
if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP)) {
11821199
val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP)
11831200
val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP)
1184-
remoteLogManager.get.resizeCacheSize(newValue)
1201+
remoteLogManager.resizeCacheSize(newValue)
11851202
info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP} updated, " +
11861203
s"old value: $oldValue, new value: $newValue")
11871204
}
11881205
if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP)) {
11891206
val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP)
11901207
val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP)
1191-
remoteLogManager.get.updateCopyQuota(newValue)
1208+
remoteLogManager.updateCopyQuota(newValue)
11921209
info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP} updated, " +
11931210
s"old value: $oldValue, new value: $newValue")
11941211
}
11951212
if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP)) {
11961213
val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP)
11971214
val newValue = newLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP)
1198-
remoteLogManager.get.updateFetchQuota(newValue)
1215+
remoteLogManager.updateFetchQuota(newValue)
11991216
info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP} updated, " +
12001217
s"old value: $oldValue, new value: $newValue")
12011218
}
1219+
1220+
val newRLMConfig = newConfig.remoteLogManagerConfig
1221+
val oldRLMConfig = oldConfig.remoteLogManagerConfig
1222+
if (newRLMConfig.remoteLogManagerCopierThreadPoolSize() != oldRLMConfig.remoteLogManagerCopierThreadPoolSize())
1223+
remoteLogManager.resizeCopierThreadPool(newRLMConfig.remoteLogManagerCopierThreadPoolSize())
1224+
1225+
if (newRLMConfig.remoteLogManagerExpirationThreadPoolSize() != oldRLMConfig.remoteLogManagerExpirationThreadPoolSize())
1226+
remoteLogManager.resizeExpirationThreadPool(newRLMConfig.remoteLogManagerExpirationThreadPoolSize())
1227+
1228+
if (newRLMConfig.remoteLogReaderThreads() != oldRLMConfig.remoteLogReaderThreads())
1229+
remoteLogManager.resizeReaderThreadPool(newRLMConfig.remoteLogReaderThreads())
12021230
}
12031231
}
12041232

@@ -1219,6 +1247,9 @@ object DynamicRemoteLogConfig {
12191247
RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP,
12201248
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP,
12211249
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP,
1222-
RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP
1250+
RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP,
1251+
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
1252+
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
1253+
RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP
12231254
)
12241255
}

‎core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,55 @@ class DynamicBrokerConfigTest {
202202
)
203203
}
204204

205+
@Test
206+
def testUpdateRemoteLogManagerDynamicThreadPool(): Unit = {
207+
val origProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
208+
val config = KafkaConfig(origProps)
209+
// When copierThreadPool and expirationThreadPool are not configured, then it defaults to the remoteLogManagerThreadPoolSize
210+
assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE, config.remoteLogManagerConfig.remoteLogManagerCopierThreadPoolSize())
211+
assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE, config.remoteLogManagerConfig.remoteLogManagerExpirationThreadPoolSize())
212+
assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_READER_THREADS, config.remoteLogManagerConfig.remoteLogReaderThreads())
213+
214+
val serverMock = mock(classOf[KafkaBroker])
215+
val remoteLogManager = mock(classOf[RemoteLogManager])
216+
when(serverMock.config).thenReturn(config)
217+
when(serverMock.remoteLogManagerOpt).thenReturn(Some(remoteLogManager))
218+
219+
config.dynamicConfig.initialize(None, None)
220+
config.dynamicConfig.addBrokerReconfigurable(new DynamicRemoteLogConfig(serverMock))
221+
222+
// Test dynamic update with valid values
223+
val props = new Properties()
224+
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP, "8")
225+
config.dynamicConfig.validate(props, perBrokerConfig = true)
226+
config.dynamicConfig.updateDefaultConfig(props)
227+
assertEquals(8, config.remoteLogManagerConfig.remoteLogManagerCopierThreadPoolSize())
228+
verify(remoteLogManager).resizeCopierThreadPool(8)
229+
230+
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP, "7")
231+
config.dynamicConfig.validate(props, perBrokerConfig = false)
232+
config.dynamicConfig.updateDefaultConfig(props)
233+
assertEquals(7, config.remoteLogManagerConfig.remoteLogManagerExpirationThreadPoolSize())
234+
verify(remoteLogManager).resizeExpirationThreadPool(7)
235+
236+
props.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, "6")
237+
config.dynamicConfig.validate(props, perBrokerConfig = true)
238+
config.dynamicConfig.updateDefaultConfig(props)
239+
assertEquals(6, config.remoteLogManagerConfig.remoteLogReaderThreads())
240+
verify(remoteLogManager).resizeReaderThreadPool(6)
241+
242+
// Test dynamic update with invalid values
243+
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP, "20")
244+
assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props, perBrokerConfig = true))
245+
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP, "15")
246+
assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props, perBrokerConfig = false))
247+
props.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, "2")
248+
assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props, perBrokerConfig = false))
249+
props.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, "0")
250+
assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(props, perBrokerConfig = true))
251+
verifyNoMoreInteractions(remoteLogManager)
252+
}
253+
205254
@nowarn("cat=deprecation")
206255
@Test
207256
def testConfigUpdateWithSomeInvalidConfigs(): Unit = {

‎storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteStorageThreadPool.java‎

Lines changed: 2 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -18,16 +18,15 @@
1818

1919
import org.apache.kafka.common.internals.FatalExitError;
2020
import org.apache.kafka.common.utils.Exit;
21+
import org.apache.kafka.common.utils.ThreadUtils;
2122
import org.apache.kafka.server.metrics.KafkaMetricsGroup;
2223

2324
import org.slf4j.Logger;
2425
import org.slf4j.LoggerFactory;
2526

2627
import java.util.concurrent.LinkedBlockingQueue;
27-
import java.util.concurrent.ThreadFactory;
2828
import java.util.concurrent.ThreadPoolExecutor;
2929
import java.util.concurrent.TimeUnit;
30-
import java.util.concurrent.atomic.AtomicInteger;
3130

3231
import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_READER_AVG_IDLE_PERCENT_METRIC;
3332
import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_READER_TASK_QUEUE_SIZE_METRIC;
@@ -41,7 +40,7 @@ public RemoteStorageThreadPool(String threadNamePrefix,
4140
int numThreads,
4241
int maxPendingTasks) {
4342
super(numThreads, numThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(maxPendingTasks),
44-
new RemoteStorageThreadFactory(threadNamePrefix));
43+
ThreadUtils.createThreadFactory(threadNamePrefix, false));
4544
metricsGroup.newGauge(REMOTE_LOG_READER_TASK_QUEUE_SIZE_METRIC.getName(),
4645
() -> getQueue().size());
4746
metricsGroup.newGauge(REMOTE_LOG_READER_AVG_IDLE_PERCENT_METRIC.getName(),
@@ -61,21 +60,6 @@ protected void afterExecute(Runnable runnable, Throwable th) {
6160
}
6261
}
6362

64-
private static class RemoteStorageThreadFactory implements ThreadFactory {
65-
private final String namePrefix;
66-
private final AtomicInteger threadNumber = new AtomicInteger(0);
67-
68-
RemoteStorageThreadFactory(String namePrefix) {
69-
this.namePrefix = namePrefix;
70-
}
71-
72-
@Override
73-
public Thread newThread(Runnable r) {
74-
return new Thread(r, namePrefix + threadNumber.getAndIncrement());
75-
}
76-
77-
}
78-
7963
public void removeMetrics() {
8064
REMOTE_STORAGE_THREAD_POOL_METRICS.forEach(metricsGroup::removeMetric);
8165
}

0 commit comments

Comments
 (0)