This is an automated email from the ASF dual-hosted git repository.
guoyangze pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 873a56361bf [FLINK-32191][netty] support set keepalive options to
NettyClient.
873a56361bf is described below
commit 873a56361bfd77c828ee743febc9dda2bb044791
Author: Weihua Hu <huweihua....@gmail.com>
AuthorDate: Tue Jun 20 20:53:30 2023 +0800
[FLINK-32191][netty] support set keepalive options to NettyClient.
---
.../generated/all_taskmanager_network_section.html | 18 +++
.../netty_shuffle_environment_configuration.html | 18 +++
.../NettyShuffleEnvironmentOptions.java | 30 +++++
.../runtime/io/network/netty/NettyClient.java | 40 ++++++
.../runtime/io/network/netty/NettyConfig.java | 13 ++
.../runtime/io/network/netty/NettyClientTest.java | 135 +++++++++++++++++++++
6 files changed, 254 insertions(+)
diff --git
a/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html
b/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html
index 7d80b82ada8..b82a5bee720 100644
--- a/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html
+++ b/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html
@@ -122,6 +122,24 @@
<td>Integer</td>
<td>The number of Netty client threads.</td>
</tr>
+ <tr>
+ <td><h5>taskmanager.network.netty.client.tcp.keepCount</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The maximum number of keepalive probes TCP should send before
Netty client dropping the connection. Note: This will not take effect when
using netty transport type of nio with an older version of JDK 8, refer to
https://bugs.openjdk.org/browse/JDK-8194298.</td>
+ </tr>
+ <tr>
+ <td><h5>taskmanager.network.netty.client.tcp.keepIdleSec</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The time (in seconds) the connection needs to remain idle
before TCP starts sending keepalive probes. Note: This will not take effect
when using netty transport type of nio with an older version of JDK 8, refer to
https://bugs.openjdk.org/browse/JDK-8194298.</td>
+ </tr>
+ <tr>
+
<td><h5>taskmanager.network.netty.client.tcp.keepIntervalSec</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The time (in seconds) between individual keepalive probes.
Note: This will not take effect when using netty transport type of nio with an
older version of JDK 8, refer to
https://bugs.openjdk.org/browse/JDK-8194298.</td>
+ </tr>
<tr>
<td><h5>taskmanager.network.netty.num-arenas</h5></td>
<td style="word-wrap: break-word;">-1</td>
diff --git
a/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html
b/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html
index cbea22beb26..9730839c0c4 100644
---
a/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html
+++
b/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html
@@ -110,6 +110,24 @@
<td>Integer</td>
<td>The number of Netty client threads.</td>
</tr>
+ <tr>
+ <td><h5>taskmanager.network.netty.client.tcp.keepCount</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The maximum number of keepalive probes TCP should send before
Netty client dropping the connection. Note: This will not take effect when
using netty transport type of nio with an older version of JDK 8, refer to
https://bugs.openjdk.org/browse/JDK-8194298.</td>
+ </tr>
+ <tr>
+ <td><h5>taskmanager.network.netty.client.tcp.keepIdleSec</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The time (in seconds) the connection needs to remain idle
before TCP starts sending keepalive probes. Note: This will not take effect
when using netty transport type of nio with an older version of JDK 8, refer to
https://bugs.openjdk.org/browse/JDK-8194298.</td>
+ </tr>
+ <tr>
+
<td><h5>taskmanager.network.netty.client.tcp.keepIntervalSec</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The time (in seconds) between individual keepalive probes.
Note: This will not take effect when using netty transport type of nio with an
older version of JDK 8, refer to
https://bugs.openjdk.org/browse/JDK-8194298.</td>
+ </tr>
<tr>
<td><h5>taskmanager.network.netty.num-arenas</h5></td>
<td style="word-wrap: break-word;">-1</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 cab12fe1972..1af9deadf35 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
@@ -491,6 +491,36 @@ public class NettyShuffleEnvironmentOptions {
+ " based on the platform. Note that the
\"epoll\" mode can get better performance, less GC and have more advanced
features which are"
+ " only available on modern Linux.");
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ public static final ConfigOption<Integer> CLIENT_TCP_KEEP_IDLE_SECONDS =
+ key("taskmanager.network.netty.client.tcp.keepIdleSec")
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The time (in seconds) the connection needs to
remain idle before TCP starts sending keepalive probes. "
+ + "Note: This will not take effect when
using netty transport type of nio with an older version of JDK 8, "
+ + "refer to
https://bugs.openjdk.org/browse/JDK-8194298.";);
+
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ public static final ConfigOption<Integer> CLIENT_TCP_KEEP_INTERVAL_SECONDS
=
+ key("taskmanager.network.netty.client.tcp.keepIntervalSec")
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The time (in seconds) between individual
keepalive probes. "
+ + "Note: This will not take effect when
using netty transport type of nio with an older version of JDK 8, "
+ + "refer to
https://bugs.openjdk.org/browse/JDK-8194298.";);
+
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ public static final ConfigOption<Integer> CLIENT_TCP_KEEP_COUNT =
+ key("taskmanager.network.netty.client.tcp.keepCount")
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The maximum number of keepalive probes TCP should
send before Netty client dropping the connection. "
+ + "Note: This will not take effect when
using netty transport type of nio with an older version of JDK 8, "
+ + "refer to
https://bugs.openjdk.org/browse/JDK-8194298.";);
+
// ------------------------------------------------------------------------
// Partition Request Options
// ------------------------------------------------------------------------
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyClient.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyClient.java
index a512fe868e8..755a8ee99e1 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyClient.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyClient.java
@@ -18,26 +18,33 @@
package org.apache.flink.runtime.io.network.netty;
+import org.apache.flink.annotation.VisibleForTesting;
+
import org.apache.flink.shaded.netty4.io.netty.bootstrap.Bootstrap;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelException;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInitializer;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelOption;
import org.apache.flink.shaded.netty4.io.netty.channel.epoll.Epoll;
+import
org.apache.flink.shaded.netty4.io.netty.channel.epoll.EpollChannelOption;
import
org.apache.flink.shaded.netty4.io.netty.channel.epoll.EpollEventLoopGroup;
import
org.apache.flink.shaded.netty4.io.netty.channel.epoll.EpollSocketChannel;
import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup;
import org.apache.flink.shaded.netty4.io.netty.channel.socket.SocketChannel;
+import
org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioChannelOption;
import
org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslHandler;
+import jdk.net.ExtendedSocketOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.net.InetSocketAddress;
+import java.net.SocketOption;
import static org.apache.flink.util.Preconditions.checkState;
@@ -45,6 +52,10 @@ class NettyClient {
private static final Logger LOG =
LoggerFactory.getLogger(NettyClient.class);
+ @VisibleForTesting static final String NIO_TCP_KEEPIDLE_KEY =
"TCP_KEEPIDLE";
+ @VisibleForTesting static final String NIO_TCP_KEEPINTERVAL_KEY =
"TCP_KEEPINTERVAL";
+ @VisibleForTesting static final String NIO_TCP_KEEPCOUNT_KEY =
"TCP_KEEPCOUNT";
+
private final NettyConfig config;
private NettyProtocol protocol;
@@ -152,6 +163,27 @@ class NettyClient {
new NioEventLoopGroup(
config.getClientNumThreads(),
NettyServer.getNamedThreadFactory(name));
bootstrap.group(nioGroup).channel(NioSocketChannel.class);
+
+ config.getTcpKeepIdleInSeconds()
+ .ifPresent(idle ->
setNioKeepaliveOptions(NIO_TCP_KEEPIDLE_KEY, idle));
+ config.getTcpKeepInternalInSeconds()
+ .ifPresent(interval ->
setNioKeepaliveOptions(NIO_TCP_KEEPINTERVAL_KEY, interval));
+ config.getTcpKeepCount()
+ .ifPresent(count ->
setNioKeepaliveOptions(NIO_TCP_KEEPCOUNT_KEY, count));
+ }
+
+ @SuppressWarnings("unchecked")
+ private void setNioKeepaliveOptions(String option, int value) {
+ try {
+ Field field = ExtendedSocketOptions.class.getField(option);
+ bootstrap.option(NioChannelOption.of((SocketOption<Integer>)
field.get(null)), value);
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ LOG.error(
+ "Ignore keepalive option {}, this may be due to using
netty transport type of nio and an older version of jdk 8,"
+ + " refer to
https://bugs.openjdk.org/browse/JDK-8194298";,
+ option,
+ e);
+ }
}
private void initEpollBootstrap() {
@@ -163,6 +195,14 @@ class NettyClient {
new EpollEventLoopGroup(
config.getClientNumThreads(),
NettyServer.getNamedThreadFactory(name));
bootstrap.group(epollGroup).channel(EpollSocketChannel.class);
+
+ config.getTcpKeepIdleInSeconds()
+ .ifPresent(idle ->
bootstrap.option(EpollChannelOption.TCP_KEEPIDLE, idle));
+ config.getTcpKeepInternalInSeconds()
+ .ifPresent(
+ interval ->
bootstrap.option(EpollChannelOption.TCP_KEEPINTVL, interval));
+ config.getTcpKeepCount()
+ .ifPresent(count ->
bootstrap.option(EpollChannelOption.TCP_KEEPCNT, count));
}
// ------------------------------------------------------------------------
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
index 979295b8d65..226b834d389 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
@@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.net.InetAddress;
+import java.util.Optional;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -142,6 +143,18 @@ public class NettyConfig {
}
}
+ public Optional<Integer> getTcpKeepIdleInSeconds() {
+ return
config.getOptional(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_IDLE_SECONDS);
+ }
+
+ public Optional<Integer> getTcpKeepInternalInSeconds() {
+ return
config.getOptional(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_INTERVAL_SECONDS);
+ }
+
+ public Optional<Integer> getTcpKeepCount() {
+ return
config.getOptional(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_COUNT);
+ }
+
@Nullable
public SSLHandlerFactory createClientSSLEngineFactory() throws Exception {
return getSSLEnabled() ?
SSLUtils.createInternalClientSSLEngineFactory(config) : null;
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyClientTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyClientTest.java
new file mode 100644
index 00000000000..6e29197ede7
--- /dev/null
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyClientTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.flink.runtime.io.network.netty;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
+import org.apache.flink.util.NetUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelOption;
+import org.apache.flink.shaded.netty4.io.netty.channel.epoll.Epoll;
+import
org.apache.flink.shaded.netty4.io.netty.channel.epoll.EpollChannelOption;
+
+import jdk.net.ExtendedSocketOptions;
+import org.junit.jupiter.api.Test;
+
+import java.net.InetAddress;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assumptions.assumeThat;
+
+/** Tests for {@link NettyClient}. */
+public class NettyClientTest {
+ @Test
+ void testSetKeepaliveOptionWithNioConfigurable() throws Exception {
+ assumeThat(keepaliveForNioConfigurable()).isTrue();
+
+ final Configuration config = new Configuration();
+ config.set(NettyShuffleEnvironmentOptions.TRANSPORT_TYPE, "nio");
+
config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_IDLE_SECONDS, 300);
+
config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_INTERVAL_SECONDS, 10);
+ config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_COUNT, 8);
+
+ try (NetUtils.Port clientPort = NetUtils.getAvailablePort()) {
+ final NettyClient client = createNettyClient(config, clientPort);
+ Map<String, Object> options =
+
client.getBootstrap().config().options().entrySet().stream()
+ .collect(Collectors.toMap(e -> e.getKey().name(),
Map.Entry::getValue));
+ assertThat(options)
+ .containsEntry(NettyClient.NIO_TCP_KEEPIDLE_KEY, 300)
+ .containsEntry(NettyClient.NIO_TCP_KEEPINTERVAL_KEY, 10)
+ .containsEntry(NettyClient.NIO_TCP_KEEPCOUNT_KEY, 8);
+ }
+ }
+
+ /**
+ * Test that keepalive options will not take effect when using netty
transport type of nio with
+ * an older version of JDK 8.
+ */
+ @Test
+ void testSetKeepaliveOptionWithNioNotConfigurable() throws Exception {
+ assumeThat(keepaliveForNioConfigurable()).isFalse();
+
+ final Configuration config = new Configuration();
+ config.set(NettyShuffleEnvironmentOptions.TRANSPORT_TYPE, "nio");
+
config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_IDLE_SECONDS, 300);
+
config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_INTERVAL_SECONDS, 10);
+ config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_COUNT, 8);
+
+ try (NetUtils.Port clientPort = NetUtils.getAvailablePort()) {
+ final NettyClient client = createNettyClient(config, clientPort);
+ Map<String, Object> options =
+
client.getBootstrap().config().options().entrySet().stream()
+ .collect(Collectors.toMap(e -> e.getKey().name(),
Map.Entry::getValue));
+ assertThat(options)
+ .doesNotContainKeys(
+ NettyClient.NIO_TCP_KEEPIDLE_KEY,
+ NettyClient.NIO_TCP_KEEPINTERVAL_KEY,
+ NettyClient.NIO_TCP_KEEPCOUNT_KEY);
+ }
+ }
+
+ @Test
+ void testSetKeepaliveOptionWithEpoll() throws Exception {
+ assumeThat(Epoll.isAvailable()).isTrue();
+
+ final Configuration config = new Configuration();
+ config.set(NettyShuffleEnvironmentOptions.TRANSPORT_TYPE, "epoll");
+
config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_IDLE_SECONDS, 300);
+
config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_INTERVAL_SECONDS, 10);
+ config.set(NettyShuffleEnvironmentOptions.CLIENT_TCP_KEEP_COUNT, 8);
+
+ try (NetUtils.Port clientPort = NetUtils.getAvailablePort()) {
+ final NettyClient client = createNettyClient(config, clientPort);
+ Map<ChannelOption<?>, Object> options =
client.getBootstrap().config().options();
+ assertThat(options)
+ .containsEntry(EpollChannelOption.TCP_KEEPIDLE, 300)
+ .containsEntry(EpollChannelOption.TCP_KEEPINTVL, 10)
+ .containsEntry(EpollChannelOption.TCP_KEEPCNT, 8);
+ }
+ }
+
+ private static boolean keepaliveForNioConfigurable() {
+ try {
+
ExtendedSocketOptions.class.getField(NettyClient.NIO_TCP_KEEPIDLE_KEY);
+ } catch (NoSuchFieldException e) {
+ return false;
+ }
+ return true;
+ }
+
+ private static NettyClient createNettyClient(Configuration config,
NetUtils.Port port)
+ throws Exception {
+
+ final NettyConfig nettyClientConfig =
+ new NettyConfig(
+ InetAddress.getLoopbackAddress(),
+ port.getPort(),
+ NettyTestUtil.DEFAULT_SEGMENT_SIZE,
+ 1,
+ config);
+
+ final NettyBufferPool bufferPool = new NettyBufferPool(1);
+ final NettyProtocol protocol = new NettyProtocol(null, null);
+
+ return NettyTestUtil.initClient(nettyClientConfig, protocol,
bufferPool);
+ }
+}
