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 b60b1e3d2a5b609aa3cdcd6fefd7be3cbdb9a612 Author: sxnan <[email protected]> AuthorDate: Mon May 27 18:24:49 2024 +0800 [FLINK-35461][config] Deprecate or update options of floating/exclusive buffer --- .../generated/all_taskmanager_network_section.html | 24 ---------- .../netty_shuffle_environment_configuration.html | 24 ---------- .../NettyShuffleEnvironmentOptions.java | 54 ++++++++++++++++++---- .../io/network/buffer/NetworkBufferPool.java | 3 +- .../NettyShuffleEnvironmentConfiguration.java | 5 +- 5 files changed, 47 insertions(+), 63 deletions(-) diff --git a/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html b/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html index 4078de37795..1cab3c95b51 100644 --- a/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html +++ b/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html @@ -74,30 +74,6 @@ <td>Integer</td> <td>The minimum difference in percentage between the newly calculated buffer size and the old one to announce the new value. Can be used to avoid constant back and forth small adjustments.</td> </tr> - <tr> - <td><h5>taskmanager.network.memory.buffers-per-channel</h5></td> - <td style="word-wrap: break-word;">2</td> - <td>Integer</td> - <td>Number of exclusive network buffers for each outgoing/incoming channel (subpartition/input channel) in the credit-based flow control model. For the outgoing channel(subpartition), this value is the effective exclusive buffers per channel. For the incoming channel(input channel), this value is the max number of exclusive buffers per channel, the number of effective exclusive network buffers per channel is dynamically calculated from taskmanager.network.memory.read-buffer.r [...] - </tr> - <tr> - <td><h5>taskmanager.network.memory.floating-buffers-per-gate</h5></td> - <td style="word-wrap: break-word;">8</td> - <td>Integer</td> - <td>Number of floating network buffers for each outgoing/incoming gate (result partition/input gate). In credit-based flow control mode, this indicates how many floating credits are shared among all the channels. The floating buffers can help relieve back-pressure caused by unbalanced data distribution among the subpartitions. For the outgoing gate(result partition), this value is the effective floating buffers per gate. For the incoming gate(input gate), this value is a reco [...] - </tr> - <tr> - <td><h5>taskmanager.network.memory.max-buffers-per-channel</h5></td> - <td style="word-wrap: break-word;">10</td> - <td>Integer</td> - <td>Number of max buffers that can be used for each channel. If a channel exceeds the number of max buffers, it will make the task become unavailable, cause the back pressure and block the data processing. This might speed up checkpoint alignment by preventing excessive growth of the buffered in-flight data in case of data skew and high number of configured floating buffers. This limit is not strictly guaranteed, and can be ignored by things like flatMap operators, records sp [...] - </tr> - <tr> - <td><h5>taskmanager.network.memory.max-overdraft-buffers-per-gate</h5></td> - <td style="word-wrap: break-word;">5</td> - <td>Integer</td> - <td>Number of max overdraft network buffers to use for each ResultPartition. The overdraft buffers will be used when the subtask cannot apply to the normal buffers due to back pressure, while subtask is performing an action that can not be interrupted in the middle, like serializing a large record, flatMap operator producing multiple records for one single input record or processing time timer producing large output. In situations like that system will allow subtask to requ [...] - </tr> <tr> <td><h5>taskmanager.network.memory.read-buffer.required-per-gate.max</h5></td> <td style="word-wrap: break-word;">(none)</td> diff --git a/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html b/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html index df099186eea..e80c283fd56 100644 --- a/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html +++ b/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html @@ -62,30 +62,6 @@ <td>Integer</td> <td>The maximum number of tpc connections between taskmanagers for data communication.</td> </tr> - <tr> - <td><h5>taskmanager.network.memory.buffers-per-channel</h5></td> - <td style="word-wrap: break-word;">2</td> - <td>Integer</td> - <td>Number of exclusive network buffers for each outgoing/incoming channel (subpartition/input channel) in the credit-based flow control model. For the outgoing channel(subpartition), this value is the effective exclusive buffers per channel. For the incoming channel(input channel), this value is the max number of exclusive buffers per channel, the number of effective exclusive network buffers per channel is dynamically calculated from taskmanager.network.memory.read-buffer.r [...] - </tr> - <tr> - <td><h5>taskmanager.network.memory.floating-buffers-per-gate</h5></td> - <td style="word-wrap: break-word;">8</td> - <td>Integer</td> - <td>Number of floating network buffers for each outgoing/incoming gate (result partition/input gate). In credit-based flow control mode, this indicates how many floating credits are shared among all the channels. The floating buffers can help relieve back-pressure caused by unbalanced data distribution among the subpartitions. For the outgoing gate(result partition), this value is the effective floating buffers per gate. For the incoming gate(input gate), this value is a reco [...] - </tr> - <tr> - <td><h5>taskmanager.network.memory.max-buffers-per-channel</h5></td> - <td style="word-wrap: break-word;">10</td> - <td>Integer</td> - <td>Number of max buffers that can be used for each channel. If a channel exceeds the number of max buffers, it will make the task become unavailable, cause the back pressure and block the data processing. This might speed up checkpoint alignment by preventing excessive growth of the buffered in-flight data in case of data skew and high number of configured floating buffers. This limit is not strictly guaranteed, and can be ignored by things like flatMap operators, records sp [...] - </tr> - <tr> - <td><h5>taskmanager.network.memory.max-overdraft-buffers-per-gate</h5></td> - <td style="word-wrap: break-word;">5</td> - <td>Integer</td> - <td>Number of max overdraft network buffers to use for each ResultPartition. The overdraft buffers will be used when the subtask cannot apply to the normal buffers due to back pressure, while subtask is performing an action that can not be interrupted in the middle, like serializing a large record, flatMap operator producing multiple records for one single input record or processing time timer producing large output. In situations like that system will allow subtask to requ [...] - </tr> <tr> <td><h5>taskmanager.network.memory.read-buffer.required-per-gate.max</h5></td> <td style="word-wrap: break-word;">(none)</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 40c69162731..ee344b3560d 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 @@ -240,8 +240,11 @@ public class NettyShuffleEnvironmentOptions { * upstream outgoing channel, max(1, configured value) will be used. In other words we ensure * that, for performance reasons, at least one buffer is used per outgoing channel regardless of * the configuration. + * + * @deprecated This option is deprecated in 1.20 and will be removed in 2.0 to simplify the + * configuration of network buffers. */ - @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) + @Deprecated public static final ConfigOption<Integer> NETWORK_BUFFERS_PER_CHANNEL = key("taskmanager.network.memory.buffers-per-channel") .intType() @@ -270,8 +273,11 @@ public class NettyShuffleEnvironmentOptions { /** * Number of floating network buffers for each outgoing/incoming gate (result partition/input * gate). + * + * @deprecated This option is deprecated in 1.20 and will be removed in 2.0 to simplify the + * configuration of network buffers. */ - @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) + @Deprecated public static final ConfigOption<Integer> NETWORK_EXTRA_BUFFERS_PER_GATE = key("taskmanager.network.memory.floating-buffers-per-gate") .intType() @@ -379,8 +385,13 @@ public class NettyShuffleEnvironmentOptions { + HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME + " is set true. "); - /** Number of max buffers can be used for each output subpartition. */ - @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) + /** + * Number of max buffers can be used for each output subpartition. + * + * @deprecated This option is deprecated in 1.20 and will be removed in 2.0 to simplify the + * configuration of network buffers. + */ + @Deprecated public static final ConfigOption<Integer> NETWORK_MAX_BUFFERS_PER_CHANNEL = key("taskmanager.network.memory.max-buffers-per-channel") .intType() @@ -393,8 +404,13 @@ public class NettyShuffleEnvironmentOptions { + " and can be ignored by things like flatMap operators, records spanning multiple buffers or single timer" + " producing large amount of data."); - /** Number of max overdraft network buffers to use for each ResultPartition. */ - @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) + /** + * Number of max overdraft network buffers to use for each ResultPartition. + * + * @deprecated This option is deprecated in 1.20 and will be removed in 2.0 to simplify the + * configuration of network buffers. + */ + @Deprecated public static final ConfigOption<Integer> NETWORK_MAX_OVERDRAFT_BUFFERS_PER_GATE = key("taskmanager.network.memory.max-overdraft-buffers-per-gate") .intType() @@ -412,9 +428,14 @@ public class NettyShuffleEnvironmentOptions { + " process any more records until the overdraft buffers are returned to the pool." + " It should be noted that this config option only takes effect for Pipelined Shuffle."); - /** The timeout for requesting exclusive buffers for each channel. */ - @Documentation.ExcludeFromDocumentation( - "This option is purely implementation related, and may be removed as the implementation changes.") + /** + * The timeout for requesting exclusive buffers for each channel. + * + * @deprecated This option is deprecated in 1.20 and will be removed in 2.0 to simplify the + * configuration of network buffers. Please use {@link + * NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_REQUEST_TIMEOUT} instead. + */ + @Deprecated public static final ConfigOption<Long> NETWORK_EXCLUSIVE_BUFFERS_REQUEST_TIMEOUT_MILLISECONDS = key("taskmanager.network.memory.exclusive-buffers-request-timeout-ms") .longType() @@ -425,6 +446,21 @@ public class NettyShuffleEnvironmentOptions { + "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks" + "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers."); + /** The timeout for requesting buffers for each channel. */ + @Documentation.ExcludeFromDocumentation( + "This option is purely implementation related, and may be removed as the implementation changes.") + public static final ConfigOption<Duration> NETWORK_BUFFERS_REQUEST_TIMEOUT = + key("taskmanager.network.memory.buffers-request-timeout") + .durationType() + .defaultValue(Duration.ofMillis(30000L)) + .withDeprecatedKeys( + "taskmanager.network.memory.exclusive-buffers-request-timeout-ms") + .withDescription( + "The timeout for requesting buffers for each channel. Since the number of maximum buffers and " + + "the number of required buffers is not the same for local buffer pools, there may be deadlock cases that the upstream" + + "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks" + + "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers."); + /** * The option to enable the new mode of hybrid shuffle. * diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java index 4cacaaa40c9..39d3697e7b9 100755 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java @@ -261,8 +261,7 @@ public class NetworkBufferPool + " or you may increase the timeout which is %dms by setting the key '%s'.", getConfigDescription(), requestSegmentsTimeout.toMillis(), - NettyShuffleEnvironmentOptions - .NETWORK_EXCLUSIVE_BUFFERS_REQUEST_TIMEOUT_MILLISECONDS + NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_REQUEST_TIMEOUT .key())); } } 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 d0cb501dd34..c8fec646ae5 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 @@ -368,10 +368,7 @@ public class NettyShuffleEnvironmentConfiguration { Collections.shuffle(shuffleDirs); Duration requestSegmentsTimeout = - Duration.ofMillis( - configuration.get( - NettyShuffleEnvironmentOptions - .NETWORK_EXCLUSIVE_BUFFERS_REQUEST_TIMEOUT_MILLISECONDS)); + configuration.get(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_REQUEST_TIMEOUT); BoundedBlockingSubpartitionType blockingSubpartitionType = getBlockingSubpartitionType(configuration);
