| /* |
| * 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.uniffle.server; |
| |
| import java.util.List; |
| |
| import org.apache.hadoop.conf.Configuration; |
| |
| import org.apache.uniffle.common.config.ConfigOption; |
| import org.apache.uniffle.common.config.ConfigOptions; |
| import org.apache.uniffle.common.config.ConfigUtils; |
| import org.apache.uniffle.common.config.RssBaseConf; |
| |
| public class ShuffleServerConf extends RssBaseConf { |
| |
| public static final String PREFIX_HADOOP_CONF = "rss.server.hadoop"; |
| public static final String SHUFFLE_SERVER_ID = "rss.server.id"; |
| |
| public static final ConfigOption<Long> SERVER_BUFFER_CAPACITY = |
| ConfigOptions.key("rss.server.buffer.capacity") |
| .longType() |
| .defaultValue(-1L) |
| .withDescription("Max memory of buffer manager for shuffle server"); |
| |
| public static final ConfigOption<Double> SERVER_BUFFER_CAPACITY_RATIO = |
| ConfigOptions.key("rss.server.buffer.capacity.ratio") |
| .doubleType() |
| .defaultValue(0.6) |
| .withDescription( |
| "JVM heap size or off-heap size(when enabling Netty) * ratio for the maximum memory of buffer manager for shuffle server, this " |
| + "is only effective when `rss.server.buffer.capacity` is not explicitly set"); |
| |
| public static final ConfigOption<Long> SERVER_READ_BUFFER_CAPACITY = |
| ConfigOptions.key("rss.server.read.buffer.capacity") |
| .longType() |
| .defaultValue(-1L) |
| .withDescription("Max size of buffer for reading data"); |
| |
| public static final ConfigOption<Double> SERVER_READ_BUFFER_CAPACITY_RATIO = |
| ConfigOptions.key("rss.server.read.buffer.capacity.ratio") |
| .doubleType() |
| .defaultValue(0.2) |
| .withDescription( |
| "JVM heap size or off-heap size(when enabling Netty) * ratio for read buffer size, this is only effective when " |
| + "`rss.server.reader.buffer.capacity.ratio` is not explicitly set"); |
| |
| public static final ConfigOption<Long> SERVER_HEARTBEAT_DELAY = |
| ConfigOptions.key("rss.server.heartbeat.delay") |
| .longType() |
| .defaultValue(10 * 1000L) |
| .withDescription("rss heartbeat initial delay ms"); |
| |
| public static final ConfigOption<Integer> SERVER_HEARTBEAT_THREAD_NUM = |
| ConfigOptions.key("rss.server.heartbeat.threadNum") |
| .intType() |
| .defaultValue(2) |
| .withDescription("rss heartbeat thread number"); |
| |
| public static final ConfigOption<Long> SERVER_HEARTBEAT_INTERVAL = |
| ConfigOptions.key("rss.server.heartbeat.interval") |
| .longType() |
| .defaultValue(10 * 1000L) |
| .withDescription("Heartbeat interval to Coordinator (ms)"); |
| |
| public static final ConfigOption<Long> SERVER_NETTY_DIRECT_MEMORY_USAGE_TRACKER_DELAY = |
| ConfigOptions.key("rss.server.netty.directMemoryTracker.memoryUsage.initialFetchDelayMs") |
| .longType() |
| .defaultValue(10 * 1000L) |
| .withDescription("Direct memory usage tracker initial delay (ms)"); |
| |
| public static final ConfigOption<Long> SERVER_NETTY_DIRECT_MEMORY_USAGE_TRACKER_INTERVAL = |
| ConfigOptions.key("rss.server.netty.directMemoryTracker.memoryUsage.updateMetricsIntervalMs") |
| .longType() |
| .defaultValue(10 * 1000L) |
| .withDescription("Direct memory usage tracker interval to MetricSystem (ms)"); |
| |
| public static final ConfigOption<Integer> SERVER_FLUSH_LOCALFILE_THREAD_POOL_SIZE = |
| ConfigOptions.key("rss.server.flush.localfile.threadPool.size") |
| .intType() |
| .defaultValue(10) |
| .withDescription("thread pool for flush data to file"); |
| |
| public static final ConfigOption<Integer> SERVER_FLUSH_HADOOP_THREAD_POOL_SIZE = |
| ConfigOptions.key("rss.server.flush.hadoop.threadPool.size") |
| .intType() |
| .defaultValue(60) |
| .withDescription("thread pool for flush data to hadoop storage"); |
| |
| public static final ConfigOption<Integer> SERVER_FLUSH_THREAD_POOL_QUEUE_SIZE = |
| ConfigOptions.key("rss.server.flush.threadPool.queue.size") |
| .intType() |
| .defaultValue(Integer.MAX_VALUE) |
| .withDescription("size of waiting queue for thread pool"); |
| |
| public static final ConfigOption<Long> SERVER_FLUSH_THREAD_ALIVE = |
| ConfigOptions.key("rss.server.flush.thread.alive") |
| .longType() |
| .defaultValue(120L) |
| .withDescription("thread idle time in pool (s)"); |
| |
| public static final ConfigOption<Long> SERVER_COMMIT_TIMEOUT = |
| ConfigOptions.key("rss.server.commit.timeout") |
| .longType() |
| .defaultValue(600000L) |
| .withDescription("Timeout when commit shuffle data (ms)"); |
| |
| public static final ConfigOption<Integer> SERVER_WRITE_RETRY_MAX = |
| ConfigOptions.key("rss.server.write.retry.max") |
| .intType() |
| .defaultValue(10) |
| .withDescription("Retry times when write fail"); |
| |
| public static final ConfigOption<Long> SERVER_APP_EXPIRED_WITHOUT_HEARTBEAT = |
| ConfigOptions.key("rss.server.app.expired.withoutHeartbeat") |
| .longType() |
| .defaultValue(60 * 1000L) |
| .withDescription( |
| "Expired time (ms) for application which has no heartbeat with coordinator"); |
| |
| public static final ConfigOption<Long> SERVER_PRE_ALLOCATION_EXPIRED = |
| ConfigOptions.key("rss.server.preAllocation.expired") |
| .longType() |
| .defaultValue(20 * 1000L) |
| .withDescription("Expired time (ms) for pre allocated buffer"); |
| |
| public static final ConfigOption<Long> SERVER_COMMIT_CHECK_INTERVAL_MAX = |
| ConfigOptions.key("rss.server.commit.check.interval.max.ms") |
| .longType() |
| .defaultValue(10000L) |
| .withDescription("Max interval(ms) for check commit status"); |
| |
| public static final ConfigOption<Long> SERVER_WRITE_SLOW_THRESHOLD = |
| ConfigOptions.key("rss.server.write.slow.threshold") |
| .longType() |
| .defaultValue(10000L) |
| .withDescription("Threshold for write slow defined"); |
| |
| public static final ConfigOption<Long> SERVER_EVENT_SIZE_THRESHOLD_L1 = |
| ConfigOptions.key("rss.server.event.size.threshold.l1") |
| .longType() |
| .defaultValue(200000L) |
| .withDescription("Threshold for event size"); |
| |
| public static final ConfigOption<Long> SERVER_EVENT_SIZE_THRESHOLD_L2 = |
| ConfigOptions.key("rss.server.event.size.threshold.l2") |
| .longType() |
| .defaultValue(1000000L) |
| .withDescription("Threshold for event size"); |
| |
| public static final ConfigOption<Long> SERVER_EVENT_SIZE_THRESHOLD_L3 = |
| ConfigOptions.key("rss.server.event.size.threshold.l3") |
| .longType() |
| .defaultValue(10000000L) |
| .withDescription("Threshold for event size"); |
| |
| public static final ConfigOption<Double> CLEANUP_THRESHOLD = |
| ConfigOptions.key("rss.server.cleanup.threshold") |
| .doubleType() |
| .checkValue( |
| ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR, |
| "clean threshold must be between 0.0 and 100.0") |
| .defaultValue(10.0) |
| .withDescription("Threshold for disk cleanup"); |
| |
| public static final ConfigOption<Double> HIGH_WATER_MARK_OF_WRITE = |
| ConfigOptions.key("rss.server.high.watermark.write") |
| .doubleType() |
| .checkValue( |
| ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR, |
| "high write watermark must be between 0.0 and 100.0") |
| .defaultValue(95.0) |
| .withDescription("If disk usage is bigger than this value, disk cannot been written"); |
| |
| public static final ConfigOption<Double> LOW_WATER_MARK_OF_WRITE = |
| ConfigOptions.key("rss.server.low.watermark.write") |
| .doubleType() |
| .checkValue( |
| ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR, |
| "low write watermark must be between 0.0 and 100.0") |
| .defaultValue(85.0) |
| .withDescription("If disk usage is smaller than this value, disk can been written again"); |
| |
| public static final ConfigOption<Boolean> DISK_CAPACITY_WATERMARK_CHECK_ENABLED = |
| ConfigOptions.key("rss.server.disk-capacity.watermark.check.enabled") |
| .booleanType() |
| .defaultValue(false) |
| .withDescription( |
| "If it is co-located with other services, the high-low watermark check " |
| + "based on the uniffle used is not correct. Due to this, the whole disk capacity " |
| + "watermark check is necessary, which will reuse the current watermark value. " |
| + "It will be disabled by default."); |
| |
| public static final ConfigOption<Long> PENDING_EVENT_TIMEOUT_SEC = |
| ConfigOptions.key("rss.server.pending.event.timeout.sec") |
| .longType() |
| .checkValue(ConfigUtils.POSITIVE_LONG_VALIDATOR, "pending event timeout must be positive") |
| .defaultValue(600L) |
| .withDescription( |
| "If disk cannot be written for timeout seconds, the flush data event will fail"); |
| |
| public static final ConfigOption<Long> DISK_CAPACITY = |
| ConfigOptions.key("rss.server.disk.capacity") |
| .longType() |
| .defaultValue(-1L) |
| .withDescription( |
| "Disk capacity that shuffle server can use. " |
| + "If it's negative, it will use the default whole space"); |
| |
| public static final ConfigOption<Double> DISK_CAPACITY_RATIO = |
| ConfigOptions.key("rss.server.disk.capacity.ratio") |
| .doubleType() |
| .defaultValue(0.9) |
| .withDescription( |
| "The maximum ratio of disk that could be used as shuffle server. This is only effective " |
| + "when `rss.server.disk.capacity` is not explicitly set"); |
| |
| public static final ConfigOption<Long> SERVER_SHUFFLE_INDEX_SIZE_HINT = |
| ConfigOptions.key("rss.server.index.size.hint") |
| .longType() |
| .defaultValue(2 * 1024L * 1024L) |
| .withDescription("The index file size hint"); |
| |
| public static final ConfigOption<Double> HEALTH_STORAGE_MAX_USAGE_PERCENTAGE = |
| ConfigOptions.key("rss.server.health.max.storage.usage.percentage") |
| .doubleType() |
| .checkValue( |
| ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR, |
| "The max usage percentage must be between 0.0 and 100.0") |
| .defaultValue(90.0) |
| .withDescription( |
| "The usage percentage of a storage exceed the value, the disk become unavailable"); |
| |
| public static final ConfigOption<Double> HEALTH_STORAGE_RECOVERY_USAGE_PERCENTAGE = |
| ConfigOptions.key("rss.server.health.storage.recovery.usage.percentage") |
| .doubleType() |
| .checkValue( |
| ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR, |
| "The recovery usage percentage must be between 0.0 and 100.0") |
| .defaultValue(80.0) |
| .withDescription( |
| "The usage percentage of an unavailable storage decline the value, the disk" |
| + " will become available"); |
| |
| public static final ConfigOption<Long> HEALTH_CHECK_INTERVAL = |
| ConfigOptions.key("rss.server.health.check.interval.ms") |
| .longType() |
| .checkValue( |
| ConfigUtils.POSITIVE_LONG_VALIDATOR, "The interval for health check must be positive") |
| .defaultValue(5000L) |
| .withDescription("The interval for health check"); |
| |
| public static final ConfigOption<Double> HEALTH_MIN_STORAGE_PERCENTAGE = |
| ConfigOptions.key("rss.server.health.min.storage.percentage") |
| .doubleType() |
| .checkValue( |
| ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR, |
| "The minimum for healthy storage percentage must be between 0.0 and 100.0") |
| .defaultValue(80.0) |
| .withDescription( |
| "The minimum fraction of storage that must pass the check mark the node as healthy"); |
| |
| public static final ConfigOption<Boolean> HEALTH_CHECK_ENABLE = |
| ConfigOptions.key("rss.server.health.check.enable") |
| .booleanType() |
| .defaultValue(false) |
| .withDescription("The switch for the health check"); |
| |
| public static final ConfigOption<List<String>> HEALTH_CHECKER_CLASS_NAMES = |
| ConfigOptions.key("rss.server.health.checker.class.names") |
| .stringType() |
| .asList() |
| .noDefaultValue() |
| .withDescription("The list of the Checker's name"); |
| |
| public static final ConfigOption<String> HEALTH_CHECKER_SCRIPT_PATH = |
| ConfigOptions.key("rss.server.health.checker.script.path") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The health script file path for HealthScriptChecker, if script file should have execute permission."); |
| |
| public static final ConfigOption<Long> HEALTH_CHECKER_SCRIPT_EXECUTE_TIMEOUT = |
| ConfigOptions.key("rss.server.health.checker.script.execute.timeout") |
| .longType() |
| .defaultValue(5000L) |
| .withDescription("The health script file execute timeout ms."); |
| |
| public static final ConfigOption<Long> HEALTH_CHECKER_LOCAL_STORAGE_EXECUTE_TIMEOUT = |
| ConfigOptions.key("rss.server.health.checker.localStorageExecutionTimeoutMS") |
| .longType() |
| .defaultValue(1000 * 60L) |
| .withDescription( |
| "The health checker for LocalStorageChecker execution timeout (Unit: ms). Default value is 1min"); |
| |
| public static final ConfigOption<Double> SERVER_MEMORY_SHUFFLE_LOWWATERMARK_PERCENTAGE = |
| ConfigOptions.key("rss.server.memory.shuffle.lowWaterMark.percentage") |
| .doubleType() |
| .checkValue( |
| ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR, |
| "The lowWaterMark for memory percentage must be between 0.0 and 100.0") |
| .defaultValue(25.0) |
| .withDescription("LowWaterMark of memory in percentage style"); |
| |
| public static final ConfigOption<Double> SERVER_MEMORY_SHUFFLE_HIGHWATERMARK_PERCENTAGE = |
| ConfigOptions.key("rss.server.memory.shuffle.highWaterMark.percentage") |
| .doubleType() |
| .checkValue( |
| ConfigUtils.PERCENTAGE_DOUBLE_VALIDATOR, |
| "The highWaterMark for memory percentage must be between 0.0 and 100.0") |
| .defaultValue(75.0) |
| .withDescription("HighWaterMark of memory in percentage style"); |
| |
| public static final ConfigOption<Long> FLUSH_COLD_STORAGE_THRESHOLD_SIZE = |
| ConfigOptions.key("rss.server.flush.cold.storage.threshold.size") |
| .longType() |
| .checkValue( |
| ConfigUtils.POSITIVE_LONG_VALIDATOR, "flush cold storage threshold must be positive") |
| .defaultValue(64L * 1024L * 1024L) |
| .withDescription( |
| "For hybrid storage, the event size exceed this value, flush data to cold storage"); |
| |
| public static final ConfigOption<String> HYBRID_STORAGE_MANAGER_SELECTOR_CLASS = |
| ConfigOptions.key("rss.server.hybrid.storage.manager.selector.class") |
| .stringType() |
| .defaultValue("org.apache.uniffle.server.storage.hybrid.DefaultStorageManagerSelector") |
| .withDescription( |
| "For hybrid storage, the storage manager selector strategy to support " |
| + "policies of flushing to different storages") |
| .withDeprecatedKeys("rss.server.multistorage.manager.selector.class"); |
| |
| public static final ConfigOption<String> HYBRID_STORAGE_FALLBACK_STRATEGY_CLASS = |
| ConfigOptions.key("rss.server.hybrid.storage.fallback.strategy.class") |
| .stringType() |
| .noDefaultValue() |
| .withDescription("For hybrid storage, fallback strategy class") |
| .withDeprecatedKeys("rss.server.multistorage.fallback.strategy.class"); |
| |
| public static final ConfigOption<Long> FALLBACK_MAX_FAIL_TIMES = |
| ConfigOptions.key("rss.server.hybrid.storage.fallback.max.fail.times") |
| .longType() |
| .defaultValue(0L) |
| .withDescription("For hybrid storage, fail times exceed the number, will switch storage") |
| .withDeprecatedKeys("rss.server.multistorage.fallback.max.fail.times"); |
| |
| public static final ConfigOption<List<String>> TAGS = |
| ConfigOptions.key("rss.server.tags") |
| .stringType() |
| .asList() |
| .noDefaultValue() |
| .withDescription("Tags list supported by shuffle server"); |
| |
| public static final ConfigOption<Long> LOCAL_STORAGE_INITIALIZE_MAX_FAIL_NUMBER = |
| ConfigOptions.key("rss.server.localstorage.initialize.max.fail.number") |
| .longType() |
| .checkValue( |
| ConfigUtils.NON_NEGATIVE_LONG_VALIDATOR, " max fail times must be non-negative") |
| .defaultValue(0L) |
| .withDescription( |
| "For localstorage, it will exit when the failed initialized local storage exceed the number"); |
| |
| public static final ConfigOption<Boolean> SINGLE_BUFFER_FLUSH_ENABLED = |
| ConfigOptions.key("rss.server.single.buffer.flush.enabled") |
| .booleanType() |
| .defaultValue(true) |
| .withDescription( |
| "Whether single buffer flush when size exceeded rss.server.single.buffer.flush.threshold"); |
| |
| public static final ConfigOption<Long> SINGLE_BUFFER_FLUSH_THRESHOLD = |
| ConfigOptions.key("rss.server.single.buffer.flush.threshold") |
| .longType() |
| .defaultValue(128 * 1024 * 1024L) |
| .withDescription("The threshold of single shuffle buffer flush"); |
| |
| public static final ConfigOption<Long> SERVER_LEAK_SHUFFLE_DATA_CHECK_INTERVAL = |
| ConfigOptions.key("rss.server.leak.shuffledata.check.interval") |
| .longType() |
| .defaultValue(3600 * 1000L) |
| .withDescription("the interval of leak shuffle data check"); |
| |
| public static final ConfigOption<Integer> SERVER_MAX_CONCURRENCY_OF_ONE_PARTITION = |
| ConfigOptions.key("rss.server.max.concurrency.of.per-partition.write") |
| .intType() |
| .defaultValue(30) |
| .withDescription( |
| "The max concurrency of single partition writer, the data partition file number is " |
| + "equal to this value. Default value is 1.") |
| .withDeprecatedKeys("rss.server.max.concurrency.of.single.partition.writer"); |
| |
| public static final ConfigOption<Integer> CLIENT_MAX_CONCURRENCY_LIMITATION_OF_ONE_PARTITION = |
| ConfigOptions.key("rss.server.max.concurrency.limit.of.per-partition.write") |
| .intType() |
| .defaultValue(Integer.MAX_VALUE) |
| .withDescription("The max concurrency limitation of per-partition writing."); |
| |
| public static final ConfigOption<Long> SERVER_TRIGGER_FLUSH_CHECK_INTERVAL = |
| ConfigOptions.key("rss.server.shuffleBufferManager.trigger.flush.interval") |
| .longType() |
| .defaultValue(0L) |
| .withDescription( |
| "The interval of trigger shuffle buffer manager to flush data to persistent storage. If <= 0" |
| + ", then this flush check would be disabled."); |
| |
| public static final ConfigOption<Long> SERVER_SHUFFLE_FLUSH_THRESHOLD = |
| ConfigOptions.key("rss.server.shuffle.flush.threshold") |
| .longType() |
| .checkValue( |
| ConfigUtils.NON_NEGATIVE_LONG_VALIDATOR, "flush threshold must be non negative") |
| .defaultValue(0L) |
| .withDescription( |
| "Threshold when flushing shuffle data to persistent storage, recommend value would be 256K, " |
| + "512K, or even 1M"); |
| |
| public static final ConfigOption<String> STORAGE_MEDIA_PROVIDER_ENV_KEY = |
| ConfigOptions.key("rss.server.storageMediaProvider.from.env.key") |
| .stringType() |
| .noDefaultValue() |
| .withDescription("The env key to get json source of local storage media provider"); |
| |
| public static final ConfigOption<Long> HUGE_PARTITION_SIZE_THRESHOLD = |
| ConfigOptions.key("rss.server.huge-partition.size.threshold") |
| .longType() |
| .defaultValue(20 * 1024 * 1024 * 1024L) |
| .withDescription( |
| "Threshold of huge partition size, once exceeding threshold, memory usage limitation and " |
| + "huge partition buffer flushing will be triggered."); |
| |
| public static final ConfigOption<Double> HUGE_PARTITION_MEMORY_USAGE_LIMITATION_RATIO = |
| ConfigOptions.key("rss.server.huge-partition.memory.limit.ratio") |
| .doubleType() |
| .defaultValue(0.2) |
| .withDescription( |
| "The memory usage limit ratio for huge partition, it will only triggered when partition's " |
| + "size exceeds the threshold of '" |
| + HUGE_PARTITION_SIZE_THRESHOLD.key() |
| + "'"); |
| |
| public static final ConfigOption<Long> SERVER_DECOMMISSION_CHECK_INTERVAL = |
| ConfigOptions.key("rss.server.decommission.check.interval") |
| .longType() |
| .checkValue(ConfigUtils.POSITIVE_LONG_VALIDATOR, "check interval times must be positive") |
| .defaultValue(60 * 1000L) |
| .withDescription( |
| "The interval(ms) to check if all applications have finish when server is decommissioning"); |
| |
| public static final ConfigOption<Boolean> SERVER_DECOMMISSION_SHUTDOWN = |
| ConfigOptions.key("rss.server.decommission.shutdown") |
| .booleanType() |
| .defaultValue(true) |
| .withDescription("Whether shutdown the server after server is decommissioned"); |
| |
| public static final ConfigOption<Integer> NETTY_SERVER_PORT = |
| ConfigOptions.key("rss.server.netty.port") |
| .intType() |
| .checkValue( |
| ConfigUtils.SERVER_PORT_VALIDATOR, |
| "check server port value is 0 or value >= 1024 && value <= 65535") |
| .defaultValue(-1) |
| .withDescription("Shuffle netty server port"); |
| |
| public static final ConfigOption<Boolean> NETTY_SERVER_EPOLL_ENABLE = |
| ConfigOptions.key("rss.server.netty.epoll.enable") |
| .booleanType() |
| .defaultValue(false) |
| .withDescription("If enable epoll model with netty server"); |
| |
| public static final ConfigOption<Integer> NETTY_SERVER_ACCEPT_THREAD = |
| ConfigOptions.key("rss.server.netty.accept.thread") |
| .intType() |
| .defaultValue(10) |
| .withDescription("Accept thread count in netty"); |
| |
| public static final ConfigOption<Integer> NETTY_SERVER_WORKER_THREAD = |
| ConfigOptions.key("rss.server.netty.worker.thread") |
| .intType() |
| .defaultValue(0) |
| .withDescription( |
| "Worker thread count in netty. When set to 0, " |
| + "the default value is dynamically set to twice the number of processor cores, " |
| + "but it will not be less than 100 to ensure the minimum throughput of the service."); |
| |
| public static final ConfigOption<Long> SERVER_NETTY_HANDLER_IDLE_TIMEOUT = |
| ConfigOptions.key("rss.server.netty.handler.idle.timeout") |
| .longType() |
| .defaultValue(60000L) |
| .withDescription("Idle timeout if there has not data"); |
| |
| public static final ConfigOption<Integer> NETTY_SERVER_CONNECT_BACKLOG = |
| ConfigOptions.key("rss.server.netty.connect.backlog") |
| .intType() |
| .defaultValue(0) |
| .withDescription( |
| "For netty server, requested maximum length of the queue of incoming connections. " |
| + "Default 0 for no backlog."); |
| |
| public static final ConfigOption<Integer> NETTY_SERVER_CONNECT_TIMEOUT = |
| ConfigOptions.key("rss.server.netty.connect.timeout") |
| .intType() |
| .defaultValue(5000) |
| .withDescription("Timeout for connection in netty"); |
| |
| public static final ConfigOption<Integer> NETTY_SERVER_SEND_BUF = |
| ConfigOptions.key("rss.server.netty.send.buf") |
| .intType() |
| .defaultValue(0) |
| .withDescription( |
| "the optimal size for send buffer(SO_SNDBUF) " |
| + "should be latency * network_bandwidth. Assuming latency = 1ms," |
| + "network_bandwidth = 10Gbps, buffer size should be ~ 1.25MB." |
| + "Default is 0, OS will dynamically adjust the buf size."); |
| |
| public static final ConfigOption<Integer> NETTY_SERVER_RECEIVE_BUF = |
| ConfigOptions.key("rss.server.netty.receive.buf") |
| .intType() |
| .defaultValue(0) |
| .withDescription( |
| "the optimal size for receive buffer(SO_RCVBUF) " |
| + "should be latency * network_bandwidth. Assuming latency = 1ms," |
| + "network_bandwidth = 10Gbps, buffer size should be ~ 1.25MB." |
| + "Default is 0, OS will dynamically adjust the buf size."); |
| |
| public static final ConfigOption<Integer> TOP_N_APP_SHUFFLE_DATA_SIZE_NUMBER = |
| ConfigOptions.key("rss.server.topN.appShuffleDataSize.number") |
| .intType() |
| .defaultValue(10) |
| .withDescription("number of topN shuffle data size of app level."); |
| |
| public static final ConfigOption<Integer> TOP_N_APP_SHUFFLE_DATA_REFRESH_INTERVAL = |
| ConfigOptions.key("rss.server.topN.appShuffleDataSize.refreshIntervalMs") |
| .intType() |
| .defaultValue(1000) |
| .withDescription( |
| "refresh interval in ms for TopN shuffle data size of app level calc task."); |
| |
| public static final ConfigOption<Integer> SUMMARY_METRIC_WAIT_QUEUE_SIZE = |
| ConfigOptions.key("rss.server.summary.metric.wait.queue.size") |
| .intType() |
| .defaultValue(1000) |
| .withDescription( |
| "size of waiting queue for thread pool that used for calc summary metric."); |
| |
| public static final ConfigOption<Integer> SUMMARY_METRIC_THREAD_POOL_CORE_SIZE = |
| ConfigOptions.key("rss.server.summary.metric.thread.pool.core.size") |
| .intType() |
| .defaultValue(2) |
| .withDescription("core thread number of thread pool that used for calc summary metric."); |
| |
| public static final ConfigOption<Integer> SUMMARY_METRIC_THREAD_POOL_MAX_SIZE = |
| ConfigOptions.key("rss.server.summary.metric.thread.pool.max.size") |
| .intType() |
| .defaultValue(20) |
| .withDescription("max thread number of thread pool that used for calc summary metric."); |
| |
| public static final ConfigOption<Integer> SUMMARY_METRIC_THREAD_POOL_KEEP_ALIVE_TIME = |
| ConfigOptions.key("rss.server.summary.metric.thread.pool.keep.alive.time") |
| .intType() |
| .defaultValue(60) |
| .withDescription( |
| "keep alive time of thread pool that used for calc summary metric, in SECONDS."); |
| |
| public ShuffleServerConf() {} |
| |
| public ShuffleServerConf(String fileName) { |
| super(); |
| boolean ret = loadConfFromFile(fileName); |
| if (!ret) { |
| throw new IllegalStateException("Fail to load config file " + fileName); |
| } |
| } |
| |
| public Configuration getHadoopConf() { |
| Configuration hadoopConf = new Configuration(); |
| for (String key : getKeySet()) { |
| if (key.startsWith(ShuffleServerConf.PREFIX_HADOOP_CONF)) { |
| String value = getString(key, ""); |
| String hadoopKey = key.substring(ShuffleServerConf.PREFIX_HADOOP_CONF.length() + 1); |
| hadoopConf.set(hadoopKey, value); |
| } |
| } |
| return hadoopConf; |
| } |
| |
| public boolean loadConfFromFile(String fileName) { |
| return loadConfFromFile(fileName, ConfigUtils.getAllConfigOptions(ShuffleServerConf.class)); |
| } |
| } |
