From b954d358ab6b960484bd0c31a172e57df1c079a2 Mon Sep 17 00:00:00 2001 From: "ezlippi.ouyang" Date: Tue, 9 Mar 2021 13:52:27 +0800 Subject: [PATCH 1/6] feat: issue2715, make netty parameter configure by system property --- .../remoting/netty/NettyClientConfig.java | 8 ++++---- .../remoting/netty/NettySystemConfig.java | 19 ++++++++++++++++++- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyClientConfig.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyClientConfig.java index fbc071b28f..5ba3534cb8 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyClientConfig.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyClientConfig.java @@ -20,23 +20,23 @@ public class NettyClientConfig { /** * Worker thread number */ - private int clientWorkerThreads = 4; + private int clientWorkerThreads = NettySystemConfig.clientWorkerSize; private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors(); private int clientOnewaySemaphoreValue = NettySystemConfig.CLIENT_ONEWAY_SEMAPHORE_VALUE; private int clientAsyncSemaphoreValue = NettySystemConfig.CLIENT_ASYNC_SEMAPHORE_VALUE; - private int connectTimeoutMillis = 3000; + private int connectTimeoutMillis = NettySystemConfig.connectTimeoutMillis; private long channelNotActiveInterval = 1000 * 60; /** * IdleStateEvent will be triggered when neither read nor write was performed for * the specified period of this time. Specify {@code 0} to disable */ - private int clientChannelMaxIdleTimeSeconds = 120; + private int clientChannelMaxIdleTimeSeconds = NettySystemConfig.clientChannelMaxIdleTimeSeconds; private int clientSocketSndBufSize = NettySystemConfig.socketSndbufSize; private int clientSocketRcvBufSize = NettySystemConfig.socketRcvbufSize; private boolean clientPooledByteBufAllocatorEnable = false; - private boolean clientCloseSocketIfTimeout = false; + private boolean clientCloseSocketIfTimeout = NettySystemConfig.clientCloseSocketIfTimeout; private boolean useTLS; diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettySystemConfig.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettySystemConfig.java index 6357c03ba8..ef767a36ed 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettySystemConfig.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettySystemConfig.java @@ -28,8 +28,17 @@ public class NettySystemConfig { "com.rocketmq.remoting.clientAsyncSemaphoreValue"; public static final String COM_ROCKETMQ_REMOTING_CLIENT_ONEWAY_SEMAPHORE_VALUE = "com.rocketmq.remoting.clientOnewaySemaphoreValue"; + public static final String COM_ROCKETMQ_REMOTING_CLIENT_WORKER_SIZE = + "com.rocketmq.remoting.client.worker.size"; + public static final String COM_ROCKETMQ_REMOTING_CLIENT_CONNECT_TIMEOUT = + "com.rocketmq.remoting.client.connect.timeout"; + public static final String COM_ROCKETMQ_REMOTING_CLIENT_CHANNEL_MAX_IDLE_SECONDS = + "com.rocketmq.remoting.client.channel.maxIdleTimeSeconds"; + public static final String COM_ROCKETMQ_REMOTING_CLIENT_CLOSE_SOCKET_IF_TIMEOUT = + "com.rocketmq.remoting.client.closeSocketIfTimeout"; - public static final boolean NETTY_POOLED_BYTE_BUF_ALLOCATOR_ENABLE = // + + public static final boolean NETTY_POOLED_BYTE_BUF_ALLOCATOR_ENABLE = // Boolean.parseBoolean(System.getProperty(COM_ROCKETMQ_REMOTING_NETTY_POOLED_BYTE_BUF_ALLOCATOR_ENABLE, "false")); public static final int CLIENT_ASYNC_SEMAPHORE_VALUE = // Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_ASYNC_SEMAPHORE_VALUE, "65535")); @@ -39,4 +48,12 @@ public class NettySystemConfig { Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE, "65535")); public static int socketRcvbufSize = Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE, "65535")); + public static int clientWorkerSize = + Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_WORKER_SIZE, "4")); + public static int connectTimeoutMillis = + Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_CONNECT_TIMEOUT, "3000")); + public static int clientChannelMaxIdleTimeSeconds = + Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_CHANNEL_MAX_IDLE_SECONDS, "120")); + public static boolean clientCloseSocketIfTimeout = + Boolean.parseBoolean(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_CLOSE_SOCKET_IF_TIMEOUT, "true")); } From 33a533b3b713ecc37f9c7168949a3e111a269ff5 Mon Sep 17 00:00:00 2001 From: "ezlippi.ouyang" Date: Fri, 12 Mar 2021 14:14:14 +0800 Subject: [PATCH 2/6] feat: add some test case --- .../remoting/netty/NettyClientConfigTest.java | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) create mode 100644 remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java new file mode 100644 index 0000000000..2352871ed3 --- /dev/null +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java @@ -0,0 +1,34 @@ +package org.apache.rocketmq.remoting.netty; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +@RunWith(MockitoJUnitRunner.class) +public class NettyClientConfigTest { + + @Test + public void testChangeConfigBySystemProperty() throws NoSuchFieldException, IllegalAccessException { + + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_WORKER_SIZE, "1"); + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_ONEWAY_SEMAPHORE_VALUE, "1023"); + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_ASYNC_SEMAPHORE_VALUE, "1024"); + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CONNECT_TIMEOUT, "2000"); + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CHANNEL_MAX_IDLE_SECONDS, "60"); + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE, "16383"); + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE, "16384"); + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CLOSE_SOCKET_IF_TIMEOUT, "false"); + + NettyClientConfig changedConfig = new NettyClientConfig(); + assertThat(changedConfig.getClientWorkerThreads()).isEqualTo(1); + assertThat(changedConfig.getClientOnewaySemaphoreValue()).isEqualTo(1023); + assertThat(changedConfig.getClientAsyncSemaphoreValue()).isEqualTo(1024); + assertThat(changedConfig.getConnectTimeoutMillis()).isEqualTo(2000); + assertThat(changedConfig.getClientChannelMaxIdleTimeSeconds()).isEqualTo(60); + assertThat(changedConfig.getClientSocketSndBufSize()).isEqualTo(16383); + assertThat(changedConfig.getClientSocketRcvBufSize()).isEqualTo(16384); + assertThat(changedConfig.isClientCloseSocketIfTimeout()).isEqualTo(false); + } +} From 03adb63c4c6ea5cedf5a2a43546eaa9cede41c74 Mon Sep 17 00:00:00 2001 From: "ezlippi.ouyang" Date: Fri, 12 Mar 2021 14:47:56 +0800 Subject: [PATCH 3/6] feat: add some test case --- .../remoting/netty/NettyClientConfigTest.java | 21 +++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java index 2352871ed3..488bd0371a 100644 --- a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java @@ -11,20 +11,33 @@ public class NettyClientConfigTest { @Test public void testChangeConfigBySystemProperty() throws NoSuchFieldException, IllegalAccessException { + System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_WORKER_SIZE, "1"); - System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_ONEWAY_SEMAPHORE_VALUE, "1023"); - System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_ASYNC_SEMAPHORE_VALUE, "1024"); System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CONNECT_TIMEOUT, "2000"); System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CHANNEL_MAX_IDLE_SECONDS, "60"); System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE, "16383"); System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE, "16384"); System.setProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CLOSE_SOCKET_IF_TIMEOUT, "false"); + + NettySystemConfig.socketSndbufSize = + Integer.parseInt(System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE, "65535")); + NettySystemConfig.socketRcvbufSize = + Integer.parseInt(System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE, "65535")); + NettySystemConfig.clientWorkerSize = + Integer.parseInt(System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_WORKER_SIZE, "4")); + NettySystemConfig.connectTimeoutMillis = + Integer.parseInt(System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CONNECT_TIMEOUT, "3000")); + NettySystemConfig.clientChannelMaxIdleTimeSeconds = + Integer.parseInt(System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CHANNEL_MAX_IDLE_SECONDS, "120")); + NettySystemConfig.clientCloseSocketIfTimeout = + Boolean.parseBoolean(System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_CLIENT_CLOSE_SOCKET_IF_TIMEOUT, "true")); + NettyClientConfig changedConfig = new NettyClientConfig(); assertThat(changedConfig.getClientWorkerThreads()).isEqualTo(1); - assertThat(changedConfig.getClientOnewaySemaphoreValue()).isEqualTo(1023); - assertThat(changedConfig.getClientAsyncSemaphoreValue()).isEqualTo(1024); + assertThat(changedConfig.getClientOnewaySemaphoreValue()).isEqualTo(65535); + assertThat(changedConfig.getClientAsyncSemaphoreValue()).isEqualTo(65535); assertThat(changedConfig.getConnectTimeoutMillis()).isEqualTo(2000); assertThat(changedConfig.getClientChannelMaxIdleTimeSeconds()).isEqualTo(60); assertThat(changedConfig.getClientSocketSndBufSize()).isEqualTo(16383); From 98237b139ae57f274c78f6765fe039995b7755eb Mon Sep 17 00:00:00 2001 From: "ezlippi.ouyang" Date: Fri, 12 Mar 2021 15:27:31 +0800 Subject: [PATCH 4/6] feat: add some test case --- .../apache/rocketmq/remoting/netty/NettyClientConfigTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java index 488bd0371a..3864ce87e0 100644 --- a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java @@ -43,5 +43,6 @@ public class NettyClientConfigTest { assertThat(changedConfig.getClientSocketSndBufSize()).isEqualTo(16383); assertThat(changedConfig.getClientSocketRcvBufSize()).isEqualTo(16384); assertThat(changedConfig.isClientCloseSocketIfTimeout()).isEqualTo(false); + } } From 67ceb64eb6ae0e59f06347bbf74004e21f4ef188 Mon Sep 17 00:00:00 2001 From: "ezlippi.ouyang" Date: Fri, 12 Mar 2021 18:05:22 +0800 Subject: [PATCH 5/6] fix: add license --- .../remoting/netty/NettyClientConfigTest.java | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java index 3864ce87e0..c0acdbe0cb 100644 --- a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java @@ -1,3 +1,20 @@ +/* + * 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.rocketmq.remoting.netty; import org.junit.Test; From 59cbbc4337d79b070ab089cb91be02af19393863 Mon Sep 17 00:00:00 2001 From: "ezlippi.ouyang" Date: Mon, 15 Mar 2021 10:08:26 +0800 Subject: [PATCH 6/6] feat: add some test case --- .../apache/rocketmq/remoting/netty/NettyClientConfigTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java index c0acdbe0cb..15cf033864 100644 --- a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyClientConfigTest.java @@ -60,6 +60,5 @@ public class NettyClientConfigTest { assertThat(changedConfig.getClientSocketSndBufSize()).isEqualTo(16383); assertThat(changedConfig.getClientSocketRcvBufSize()).isEqualTo(16384); assertThat(changedConfig.isClientCloseSocketIfTimeout()).isEqualTo(false); - } }