This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 312df35d00bb0823749df913820561e01f9bcf2d Author: sxnan <[email protected]> AuthorDate: Tue May 28 09:22:26 2024 +0800 [FLINK-35461][config] Deprecate or update options of shuffle compression --- .../generated/all_taskmanager_network_section.html | 8 +------- .../netty_shuffle_environment_configuration.html | 8 +------- .../configuration/NettyShuffleEnvironmentOptions.java | 11 ++++++++--- .../NettyShuffleEnvironmentConfiguration.java | 19 +++++++++++++++++-- 4 files changed, 27 insertions(+), 19 deletions(-) diff --git a/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html b/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html index 1cab3c95b51..46090128563 100644 --- a/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html +++ b/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html @@ -8,17 +8,11 @@ </tr> </thead> <tbody> - <tr> - <td><h5>taskmanager.network.batch-shuffle.compression.enabled</h5></td> - <td style="word-wrap: break-word;">true</td> - <td>Boolean</td> - <td>Boolean flag indicating whether the shuffle data will be compressed for batch shuffle mode. Note that data is compressed per buffer and compression can incur extra CPU overhead, so it is more effective for IO bounded scenario when compression ratio is high.</td> - </tr> <tr> <td><h5>taskmanager.network.compression.codec</h5></td> <td style="word-wrap: break-word;">LZ4</td> <td><p>Enum</p></td> - <td>The codec to be used when compressing shuffle data, only "LZ4", "LZO" and "ZSTD" are supported now. Through tpc-ds test of these three algorithms, the results show that "LZ4" algorithm has the highest compression and decompression speed, but the compression ratio is the lowest. "ZSTD" has the highest compression ratio, but the compression and decompression speed is the slowest, and LZO is between the two. Also note that this option is experimental and might be changed in [...] + <td>The codec to be used when compressing shuffle data. If it is "NONE", compression is disable. If it is not "NONE", only "LZ4", "LZO" and "ZSTD" are supported now. Through tpc-ds test of these three algorithms, the results show that "LZ4" algorithm has the highest compression and decompression speed, but the compression ratio is the lowest. "ZSTD" has the highest compression ratio, but the compression and decompression speed is the slowest, and LZO is between the two. Also [...] </tr> <tr> <td><h5>taskmanager.network.detailed-metrics</h5></td> diff --git a/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html b/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html index e80c283fd56..e9446edc23a 100644 --- a/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html +++ b/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html @@ -26,17 +26,11 @@ <td>Boolean</td> <td>Enable SSL support for the taskmanager data transport. This is applicable only when the global flag for internal SSL (security.ssl.internal.enabled) is set to true</td> </tr> - <tr> - <td><h5>taskmanager.network.batch-shuffle.compression.enabled</h5></td> - <td style="word-wrap: break-word;">true</td> - <td>Boolean</td> - <td>Boolean flag indicating whether the shuffle data will be compressed for batch shuffle mode. Note that data is compressed per buffer and compression can incur extra CPU overhead, so it is more effective for IO bounded scenario when compression ratio is high.</td> - </tr> <tr> <td><h5>taskmanager.network.compression.codec</h5></td> <td style="word-wrap: break-word;">LZ4</td> <td><p>Enum</p></td> - <td>The codec to be used when compressing shuffle data, only "LZ4", "LZO" and "ZSTD" are supported now. Through tpc-ds test of these three algorithms, the results show that "LZ4" algorithm has the highest compression and decompression speed, but the compression ratio is the lowest. "ZSTD" has the highest compression ratio, but the compression and decompression speed is the slowest, and LZO is between the two. Also note that this option is experimental and might be changed in [...] + <td>The codec to be used when compressing shuffle data. If it is "NONE", compression is disable. If it is not "NONE", only "LZ4", "LZO" and "ZSTD" are supported now. Through tpc-ds test of these three algorithms, the results show that "LZ4" algorithm has the highest compression and decompression speed, but the compression ratio is the lowest. "ZSTD" has the highest compression ratio, but the compression and decompression speed is the slowest, and LZO is between the two. Also [...] </tr> <tr> <td><h5>taskmanager.network.detailed-metrics</h5></td> diff --git a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java index ee344b3560d..af08b80a3ad 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java @@ -90,8 +90,12 @@ public class NettyShuffleEnvironmentOptions { * * <p>Note: Data is compressed per buffer and compression can incur extra CPU overhead so it is * more effective for IO bounded scenario when data compression ratio is high. + * + * @deprecated This option is deprecated in 1.20 and will be removed in 2.0. Please set the + * {@link NettyShuffleEnvironmentOptions#SHUFFLE_COMPRESSION_CODEC} to NONE to disable the + * compression. */ - @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) + @Deprecated public static final ConfigOption<Boolean> BATCH_SHUFFLE_COMPRESSION_ENABLED = key("taskmanager.network.batch-shuffle.compression.enabled") .booleanType() @@ -106,13 +110,13 @@ public class NettyShuffleEnvironmentOptions { /** The codec to be used when compressing shuffle data. */ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) - @Experimental public static final ConfigOption<CompressionCodec> SHUFFLE_COMPRESSION_CODEC = key("taskmanager.network.compression.codec") .enumType(CompressionCodec.class) .defaultValue(CompressionCodec.LZ4) .withDescription( - "The codec to be used when compressing shuffle data, only \"LZ4\", \"LZO\" " + "The codec to be used when compressing shuffle data. If it is \"NONE\", " + + "compression is disable. If it is not \"NONE\", only \"LZ4\", \"LZO\" " + "and \"ZSTD\" are supported now. Through tpc-ds test of these " + "three algorithms, the results show that \"LZ4\" algorithm has " + "the highest compression and decompression speed, but the " @@ -123,6 +127,7 @@ public class NettyShuffleEnvironmentOptions { /** Supported compression codec. */ public enum CompressionCodec { + NONE, LZ4, LZO, ZSTD diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java index c8fec646ae5..710f2bbbf7e 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java @@ -373,11 +373,26 @@ public class NettyShuffleEnvironmentConfiguration { BoundedBlockingSubpartitionType blockingSubpartitionType = getBlockingSubpartitionType(configuration); - boolean batchShuffleCompressionEnabled = - configuration.get(NettyShuffleEnvironmentOptions.BATCH_SHUFFLE_COMPRESSION_ENABLED); CompressionCodec compressionCodec = configuration.get(NettyShuffleEnvironmentOptions.SHUFFLE_COMPRESSION_CODEC); + boolean batchShuffleCompressionEnabled; + if (compressionCodec == CompressionCodec.NONE) { + batchShuffleCompressionEnabled = false; + } else { + batchShuffleCompressionEnabled = + configuration.get( + NettyShuffleEnvironmentOptions.BATCH_SHUFFLE_COMPRESSION_ENABLED); + + if (!batchShuffleCompressionEnabled) { + LOG.warn( + "Deprecated configuration key {} is used to disable the compression. " + + "Please set the {} to \"None\" to disable the compression.", + NettyShuffleEnvironmentOptions.BATCH_SHUFFLE_COMPRESSION_ENABLED.key(), + NettyShuffleEnvironmentOptions.SHUFFLE_COMPRESSION_CODEC.key()); + } + } + int maxNumConnections = Math.max( 1,
