Skip to content

[server] Add support of dynamic historical lookup threads and max queued requests reconfiguration #4034

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

Open
zuston wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[server] Add support of dynamic historical lookup threads and max queued requests reconfiguration #4034

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
4 changes: 4 additions & 0 deletions fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,13 +51,15 @@
import static org.apache.fluss.config.ConfigOptions.KV_SNAPSHOT_INTERVAL;
import static org.apache.fluss.config.ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER;
import static org.apache.fluss.config.ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED;
import static org.apache.fluss.config.ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_STRATEGY;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS;
import static org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO;
import static org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO;
import static org.apache.fluss.config.ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS;
import static org.apache.fluss.config.ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO;
import static org.apache.fluss.config.ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE;
import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_CREDENTIALS;
import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG;
import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock;
Expand Down Expand Up @@ -86,6 +88,8 @@ class DynamicServerConfig {
SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO.key(),
SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS.key(),
SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key(),
NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key(),
// Config options for remote.data.dirs
REMOTE_DATA_DIRS.key(),
REMOTE_DATA_DIRS_STRATEGY.key(),
Expand Down
17 changes: 16 additions & 1 deletion ...c/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,11 +24,18 @@

import java.time.Duration;

/** Validates dynamic historical lookup cache settings. */
/** Validates dynamic historical lookup settings. */
final class HistoricalLookupCacheConfigValidator implements ServerReconfigurable {

@Override
public void validate(Configuration newConfig) throws ConfigException {
validatePositive(
newConfig.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE),
ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key());
validatePositive(
newConfig.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS),
ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key());

double newMaxRatio =
newConfig.get(
ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO);
Expand Down Expand Up @@ -56,4 +63,12 @@ public void validate(Configuration newConfig) throws ConfigException {

@Override
public void reconfigure(Configuration newConfig) {}

private static void validatePositive(int value, String configKey) throws ConfigException {
if (value <= 0) {
throw new ConfigException(
String.format(
"Invalid configuration for %s, it must be greater than 0.", configKey));
}
}
}
102 changes: 92 additions & 10 deletions fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.TableConfig;
import org.apache.fluss.exception.ConfigException;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.HistoricalPartitionThrottledException;
import org.apache.fluss.exception.InvalidPartitionException;
Expand Down Expand Up @@ -71,7 +72,6 @@
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
Expand Down Expand Up @@ -131,12 +131,13 @@ class HistoricalLakeLookupManager implements AutoCloseable {
private volatile long lakeConfigVersion;
private final @Nullable PluginManager pluginManager;
private final Counter capacityEvictions;
private final int maxQueuedHistoricalRequests;
private final Semaphore lookupPermits;
private volatile int maxQueuedHistoricalRequests;
private final AdjustableSemaphore lookupPermits;
// Accepted lookup futures tracked so close() can cancel tasks left after executor shutdown.
private final Set<CompletableFuture<LookupResultForBucket>> pendingLookups;
private final Cache<Long, CachedLakeTableLookuper> lakeTableLookupers;
private final ExecutorService historicalPartitionExecutor;
private final ThreadPoolExecutor historicalPartitionExecutor;
private volatile int maxThreadPoolSize;
private final File historicalLookupCacheRootDir;
private final long dataDirVolumeBytes;
// TODO: Introduce a minimum lookup cache disk ratio (default 0.01). When disk usage is high,
Expand Down Expand Up @@ -173,7 +174,7 @@ class HistoricalLakeLookupManager implements AutoCloseable {
HistoricalLakeLookupManager(
Configuration conf,
@Nullable PluginManager pluginManager,
@Nullable ExecutorService historicalPartitionExecutor,
@Nullable ThreadPoolExecutor historicalPartitionExecutor,
File dataDir,
long dataDirVolumeBytes,
Ticker ticker,
Expand Down Expand Up @@ -209,6 +210,7 @@ class HistoricalLakeLookupManager implements AutoCloseable {
historicalPartitionExecutor == null
? createHistoricalPartitionExecutor(maxThreadPoolSize)
: historicalPartitionExecutor;
this.maxThreadPoolSize = maxThreadPoolSize;
this.lakeTableLookupers =
Caffeine.newBuilder()
.maximumSize(MAX_CACHED_TABLES)
Expand All @@ -221,7 +223,7 @@ class HistoricalLakeLookupManager implements AutoCloseable {
.executor(Runnable::run)
.removalListener(this::onLookuperRemoved)
.build();
this.lookupPermits = new Semaphore(maxQueuedHistoricalRequests);
this.lookupPermits = new AdjustableSemaphore(maxQueuedHistoricalRequests);
this.pendingLookups = ConcurrentHashMap.newKeySet();
}

Expand Down Expand Up @@ -341,7 +343,7 @@ private CompletableFuture<LookupResultForBucket> submitLookup(
return future;
}

private ExecutorService createHistoricalPartitionExecutor(int maxThreadPoolSize) {
private ThreadPoolExecutor createHistoricalPartitionExecutor(int maxThreadPoolSize) {
ThreadPoolExecutor executor =
new ThreadPoolExecutor(
maxThreadPoolSize,
Expand All @@ -354,6 +356,18 @@ private ExecutorService createHistoricalPartitionExecutor(int maxThreadPoolSize)
return executor;
}

private static final class AdjustableSemaphore extends Semaphore {
private static final long serialVersionUID = 1L;

private AdjustableSemaphore(int permits) {
super(permits);
}

private void decreasePermits(int reduction) {
super.reducePermits(reduction);
}
}

/** Invalidates the cached lake lookuper for the given table. */
void invalidateTableLookuper(long tableId) {
lakeTableLookupers.invalidate(tableId);
Expand All @@ -369,17 +383,37 @@ Counter capacityEvictions() {
return capacityEvictions;
}

/** Returns the number of accepted historical lookup requests that have not completed. */
/**
* Returns the number of accepted historical lookup requests that have not completed.
*
* <p>This value is intended for monitoring and may be transient while requests complete or the
* request limit is reconfigured.
*/
int numInflightRequests() {
return maxQueuedHistoricalRequests - lookupPermits.availablePermits();
}

/** Validates the historical lookup settings that can be changed at runtime. */
static void validateConfig(Configuration config) throws ConfigException {
validatePositive(
config.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS),
ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key());
validatePositive(
config.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE),
ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key());
}

/** Applies dynamic historical lookup configuration changes. */
void reconfigure(Configuration newConf) {
void reconfigure(Configuration newConf) throws ConfigException {
checkNotNull(newConf, "newConf must not be null.");
validateConfig(newConf);
boolean lakeConfigChanged;
boolean cacheLimitChanged;
boolean expirationChanged;
int newMaxQueuedHistoricalRequests =
newConf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS);
int newMaxThreadPoolSize =
newConf.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE);
Duration newExpiration =
newConf.get(
ConfigOptions
Expand All @@ -391,8 +425,28 @@ void reconfigure(Configuration newConf) {
ConfigOptions
.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO));
cacheLimitChanged = newMaxBytesPerTable != lookupCacheMaxDiskBytesPerTable;
lookupCacheMaxDiskBytesPerTable = newMaxBytesPerTable;

if (newMaxThreadPoolSize != maxThreadPoolSize) {
resizeHistoricalPartitionThreadPool(newMaxThreadPoolSize);
LOG.info(
"{} reconfigured: {} -> {}",
ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key(),
maxThreadPoolSize,
newMaxThreadPoolSize);
maxThreadPoolSize = newMaxThreadPoolSize;
}

if (newMaxQueuedHistoricalRequests != maxQueuedHistoricalRequests) {
adjustLookupPermits(newMaxQueuedHistoricalRequests);
LOG.info(
"{} reconfigured: {} -> {}",
ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key(),
maxQueuedHistoricalRequests,
newMaxQueuedHistoricalRequests);
maxQueuedHistoricalRequests = newMaxQueuedHistoricalRequests;
}

lookupCacheMaxDiskBytesPerTable = newMaxBytesPerTable;
lakeConfigChanged = hasLakeConfigChanged(conf, newConf);
expirationChanged =
!newExpiration.equals(
Expand Down Expand Up @@ -424,6 +478,34 @@ void reconfigure(Configuration newConf) {
}
}

private static void validatePositive(int value, String configKey) throws ConfigException {
if (value <= 0) {
throw new ConfigException(
String.format(
"Invalid configuration for %s, it must be greater than 0.", configKey));
}
}

private void resizeHistoricalPartitionThreadPool(int newMaxThreadPoolSize) {
int currentMaxThreadPoolSize = historicalPartitionExecutor.getMaximumPoolSize();
if (newMaxThreadPoolSize > currentMaxThreadPoolSize) {
historicalPartitionExecutor.setMaximumPoolSize(newMaxThreadPoolSize);
historicalPartitionExecutor.setCorePoolSize(newMaxThreadPoolSize);
} else if (newMaxThreadPoolSize < currentMaxThreadPoolSize) {
historicalPartitionExecutor.setCorePoolSize(newMaxThreadPoolSize);
historicalPartitionExecutor.setMaximumPoolSize(newMaxThreadPoolSize);
}
}

private void adjustLookupPermits(int newMaxQueuedHistoricalRequests) {
int permitDelta = newMaxQueuedHistoricalRequests - maxQueuedHistoricalRequests;
if (permitDelta > 0) {
lookupPermits.release(permitDelta);
} else if (permitDelta < 0) {
lookupPermits.decreasePermits(-permitDelta);
}
}

private LookupResultForBucket lookupInternal(
LookupDataForBucket lookupData,
TableInfo tableInfo,
Expand Down
1 change: 1 addition & 0 deletions fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -414,6 +414,7 @@ public int getCoordinatorEpoch() {
public void validate(Configuration newConfig) throws ConfigException {
// Type validation is already handled by DynamicServerConfig.
// Here we only do basic sanity checks.
HistoricalLakeLookupManager.validateConfig(newConfig);
int newMinInSyncReplicas =
newConfig.get(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER);
if (newMinInSyncReplicas <= 0) {
Expand Down
24 changes: 24 additions & 0 deletions fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -429,6 +429,30 @@ void testReStartupContainsNoMatchedDynamicConfig() throws Exception {
}
}

@Test
void testHistoricalLookupConfigsCanBeChangedDynamically() throws Exception {
DynamicConfigManager dynamicConfigManager = createManager(new Configuration());
dynamicConfigManager.startup();

dynamicConfigManager.alterConfigs(
Arrays.asList(
new AlterConfig(
ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE
.key(),
"12",
AlterConfigOpType.SET),
new AlterConfig(
ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key(),
"60",
AlterConfigOpType.SET)));

assertThat(zookeeperClient.fetchEntityConfig())
.containsEntry(
ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key(), "12")
.containsEntry(
ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key(), "60");
}

@Test
void testPreventInvalidConfig() throws Exception {
// Test that generic type validation prevents invalid config values
Expand Down
Loading
Loading