This is an automated email from the ASF dual-hosted git repository.
gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 1c9fd697d [INLONG-3644][TubeMQ] Upgrade netty version and tidy up
other dependencies (#3655)
1c9fd697d is described below
commit 1c9fd697defffd4149b764799f2a9c5ad65c6cc0
Author: baomingyu <[email protected]>
AuthorDate: Wed Apr 13 20:48:36 2022 +0800
[INLONG-3644][TubeMQ] Upgrade netty version and tidy up other dependencies
(#3655)
* #3644 Upgrade netty version and tidy up dependencies.
* hanlde netty exception and different log configuration
---
inlong-manager/manager-common/pom.xml | 2 +-
inlong-manager/manager-service/pom.xml | 2 +-
inlong-tubemq/bin/env.cmd | 6 +-
inlong-tubemq/bin/env.sh | 8 +-
inlong-tubemq/conf/log4j.properties | 46 ----
inlong-tubemq/conf/log4j2.xml | 126 +++++++++++
inlong-tubemq/conf/master.log4j.properties | 25 ---
inlong-tubemq/conf/tools.log4j.properties | 24 --
inlong-tubemq/pom.xml | 220 -------------------
inlong-tubemq/tubemq-client/pom.xml | 9 +-
.../tubemq/client/consumer/RmtDataCache.java | 8 +-
inlong-tubemq/tubemq-connectors/pom.xml | 2 +
.../tubemq-connector-flink/pom.xml | 3 +-
.../tubemq-connector-spark/pom.xml | 3 +-
inlong-tubemq/tubemq-core/pom.xml | 35 ++-
.../inlong/tubemq/corebase/utils/AddressUtils.java | 4 +-
.../apache/inlong/tubemq/corerpc/RpcConstants.java | 3 +
.../inlong/tubemq/corerpc/netty/EventLoopUtil.java | 115 ++++++++++
.../inlong/tubemq/corerpc/netty/NettyClient.java | 55 ++---
.../tubemq/corerpc/netty/NettyClientFactory.java | 120 +++++-----
.../tubemq/corerpc/netty/NettyProtocolDecoder.java | 90 ++++----
.../tubemq/corerpc/netty/NettyProtocolEncoder.java | 45 ++--
.../tubemq/corerpc/netty/NettyRequestContext.java | 20 +-
.../tubemq/corerpc/netty/NettyRpcServer.java | 113 +++++-----
.../tubemq/corerpc/netty/ReadTimeoutHandler.java | 244 ---------------------
.../corerpc/netty/NettyProtocolEncoderTest.java | 9 +-
inlong-tubemq/tubemq-example/pom.xml | 1 +
inlong-tubemq/tubemq-server/pom.xml | 47 +++-
pom.xml | 107 ++++++++-
29 files changed, 668 insertions(+), 824 deletions(-)
diff --git a/inlong-manager/manager-common/pom.xml
b/inlong-manager/manager-common/pom.xml
index 8beba780b..23899108a 100644
--- a/inlong-manager/manager-common/pom.xml
+++ b/inlong-manager/manager-common/pom.xml
@@ -134,7 +134,7 @@
</dependency>
<dependency>
<groupId>javax.servlet</groupId>
- <artifactId>servlet-api</artifactId>
+ <artifactId>javax.servlet-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
diff --git a/inlong-manager/manager-service/pom.xml
b/inlong-manager/manager-service/pom.xml
index 07c8197d3..eb536e1b4 100644
--- a/inlong-manager/manager-service/pom.xml
+++ b/inlong-manager/manager-service/pom.xml
@@ -66,7 +66,7 @@
<dependency>
<groupId>javax.servlet</groupId>
- <artifactId>servlet-api</artifactId>
+ <artifactId>javax.servlet-api</artifactId>
<scope>provided</scope>
</dependency>
diff --git a/inlong-tubemq/bin/env.cmd b/inlong-tubemq/bin/env.cmd
index 513c664e5..a12689f50 100644
--- a/inlong-tubemq/bin/env.cmd
+++ b/inlong-tubemq/bin/env.cmd
@@ -22,7 +22,7 @@ REM Java runtime evironment could be specified here.
set BASE_DIR=%~dp0..
set CLASSPATH=%BASE_DIR%\lib\*;%BASE_DIR%\tubemq-server\target\*;%CLASSPATH%
-set GENERIC_ARGS="-Dtubemq.home=%BASE_DIR%" -cp "%CLASSPATH%"
"-Dlog4j.configuration=file:%BASE_DIR%\conf\master.log4j.properties"
+set GENERIC_ARGS="-Dtubemq.home=%BASE_DIR%" -cp "%CLASSPATH%"
"-Dtubemq.log.path=%BASE_DIR%\logs"
"-Dlog4j.configurationFile=%BASE_DIR%\conf\log4j2.xml"
REM If there's no system-wide JAVA_HOME or there's need to run on specific
Java,
REM please uncomment the following JAVA_HOME line, and specify the java home
path.
@@ -31,5 +31,5 @@ REM set JAVA_HOME=C:\Program Files\Java\jdk1.8.0_241
set JAVA="%JAVA_HOME%\bin\java"
REM One may add extra Java runtime flags in addition to each role: Master or
Broker
-set MASTER_JVM_OPTS=-Xmx1g -Xms256m -server
-set BROKER_JVM_OPTS=-Xmx1g -Xms512m -server
\ No newline at end of file
+set MASTER_JVM_OPTS=-Xmx1g -Xms256m -server "-Dtubemq.log.prefix=master"
+set BROKER_JVM_OPTS=-Xmx1g -Xms512m -server "-Dtubemq.log.prefix=broker"
\ No newline at end of file
diff --git a/inlong-tubemq/bin/env.sh b/inlong-tubemq/bin/env.sh
index 2f04da4b6..aabac44f8 100755
--- a/inlong-tubemq/bin/env.sh
+++ b/inlong-tubemq/bin/env.sh
@@ -45,19 +45,19 @@ TOOLS_JVM_ARGS="-Xmx512m -Xms512m
-Dtubemq.home=$tubemq_home -cp $CLASSPATH "
TOOL_REPAIR_JVM_ARGS="-Xmx24g -Xms8g -Dtubemq.home=$tubemq_home -cp $CLASSPATH
"
if [ -z "$MASTER_ARGS" ]; then
- export MASTER_ARGS="$MASTER_JVM_ARGS
-Dlog4j.configuration=file:$BASE_DIR/conf/master.log4j.properties"
+ export MASTER_ARGS="$MASTER_JVM_ARGS -Dtubemq.log.prefix=master
-Dtubemq.log.path=$LOG_DIR
-Dlog4j.configurationFile=${BASE_DIR}/conf/log4j2.xml"
fi
if [ -z "$BROKER_ARGS" ]; then
- export BROKER_ARGS="$BROKER_JVM_ARGS
-Dlog4j.configuration=file:$BASE_DIR/conf/log4j.properties"
+ export BROKER_ARGS="$BROKER_JVM_ARGS -Dtubemq.log.prefix=broker
-Dtubemq.log.path=$LOG_DIR
-Dlog4j.configurationFile=${BASE_DIR}/conf/log4j2.xml"
fi
if [ -z "$TOOLS_ARGS" ]; then
- export TOOLS_ARGS="$TOOLS_JVM_ARGS
-Dlog4j.configuration=file:$BASE_DIR/conf/tools.log4j.properties"
+ export TOOLS_ARGS="$TOOLS_JVM_ARGS -Dtubemq.log.prefix=tools
-Dtubemq.log.path=$LOG_DIR
-Dlog4j.configurationFile=${BASE_DIR}/conf/log4j2.xml"
fi
if [ -z "$TOOL_REPAIR_ARGS" ]; then
- export TOOL_REPAIR_ARGS="$TOOL_REPAIR_JVM_ARGS
-Dlog4j.configuration=file:$BASE_DIR/conf/tools.log4j.properties"
+ export TOOL_REPAIR_ARGS="$TOOL_REPAIR_JVM_ARGS -Dtubemq.log.prefix=tools
-Dtubemq.log.path=$LOG_DIR
-Dlog4j.configurationFile=${BASE_DIR}/conf/log4j2.xml"
fi
diff --git a/inlong-tubemq/conf/log4j.properties
b/inlong-tubemq/conf/log4j.properties
deleted file mode 100644
index 2f53338f6..000000000
--- a/inlong-tubemq/conf/log4j.properties
+++ /dev/null
@@ -1,46 +0,0 @@
-#
-# 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.
-#
-
-log4j.rootLogger=INFO, ServerDailyRollingFile
-log4j.appender.ServerDailyRollingFile=org.apache.log4j.DailyRollingFileAppender
-log4j.appender.ServerDailyRollingFile.DatePattern='.'yyyy-MM-dd
-log4j.appender.ServerDailyRollingFile.File=${tubemq.home}/logs/broker.log
-log4j.appender.ServerDailyRollingFile.layout=org.apache.log4j.PatternLayout
-log4j.appender.ServerDailyRollingFile.layout.ConversionPattern=[%p]
%d{yyyy-MM-dd HH:mm:ss,SSS} [%c{1}] %m%n
-log4j.appender.ServerDailyRollingFile.Append=true
-log4j.logger.PutCounterGroup=INFO,PutCounterGroup
-log4j.additivity.PutCounterGroup=false
-log4j.appender.PutCounterGroup=org.apache.log4j.RollingFileAppender
-log4j.appender.PutCounterGroup.MaxFileSize=200MB
-log4j.appender.PutCounterGroup.MaxBackupIndex=20
-log4j.appender.PutCounterGroup.BufferedIO=false
-log4j.appender.PutCounterGroup.BufferSize=8192
-log4j.appender.PutCounterGroup.File=${tubemq.home}/logs/put_transfer.log
-log4j.appender.PutCounterGroup.layout=org.apache.log4j.PatternLayout
-log4j.appender.PutCounterGroup.layout.ConversionPattern=%d{yyyy-MM-dd
HH:mm:ss} %m%n
-log4j.appender.PutCounterGroup.Append=true
-log4j.logger.GetCounterGroup=INFO,GetCounterGroup
-log4j.additivity.GetCounterGroup=false
-log4j.appender.GetCounterGroup=org.apache.log4j.RollingFileAppender
-log4j.appender.GetCounterGroup.MaxFileSize=200MB
-log4j.appender.GetCounterGroup.MaxBackupIndex=20
-log4j.appender.GetCounterGroup.BufferedIO=false
-log4j.appender.GetCounterGroup.BufferSize=8192
-log4j.appender.GetCounterGroup.File=${tubemq.home}/logs/get_transfer.log
-log4j.appender.GetCounterGroup.layout=org.apache.log4j.PatternLayout
-log4j.appender.GetCounterGroup.layout.ConversionPattern=%d{yyyy-MM-dd
HH:mm:ss} %m%n
-log4j.appender.GetCounterGroup.Append=true
diff --git a/inlong-tubemq/conf/log4j2.xml b/inlong-tubemq/conf/log4j2.xml
new file mode 100644
index 000000000..4440c84ac
--- /dev/null
+++ b/inlong-tubemq/conf/log4j2.xml
@@ -0,0 +1,126 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<configuration status="WARN" monitorInterval="30">
+ <Properties>
+ <property name="basePath">${sys:tubemq.log.path}</property>
+ <property name="filePrefix">${sys:tubemq.log.prefix}</property>
+ <property name="log_pattern">%d{yyyy-MM-dd HH:mm:ss.SSS} -%5p ${PID:-}
[%15.15t] %-30.30C{1.} : %m%n</property>
+ <property name="every_file_size">1G</property>
+ <property name="output_log_level">DEBUG</property>
+ <property name="rolling_max">50</property>
+ <property
name="info_fileName">${basePath}/${filePrefix}_info.log</property>
+ <property
name="info_filePattern">${basePath}/${filePrefix}_info-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="info_max">10</property>
+ <property
name="debug_fileName">${basePath}/${filePrefix}_debug.log</property>
+ <property
name="debug_filePattern">${basePath}/${filePrefix}_debug-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="debug_max">10</property>
+ <property
name="warn_fileName">${basePath}/${filePrefix}_warn.log</property>
+ <property
name="warn_filePattern">${basePath}/${filePrefix}_warn-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="warn_max">10</property>
+ <property
name="error_fileName">${basePath}/${filePrefix}_error.log</property>
+ <property
name="error_filePattern">${basePath}/${filePrefix}_error-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="error_max">10</property>
+ <property
name="put_transfer_fileName">${basePath}/${filePrefix}_put_transfer.log</property>
+ <property
name="put_transfer_filePattern">${basePath}/${filePrefix}_put_transfer-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="put_transfer_max">10</property>
+ <property
name="get_transfer_fileName">${basePath}/${filePrefix}_get_transfer.log</property>
+ <property
name="get_transfer_filePattern">${basePath}/${filePrefix}_get_transfer-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="get_transfer_max">10</property>
+ <property name="console_print_level">DEBUG</property>
+ </Properties>
+
+ <appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <ThresholdFilter level="${console_print_level}" onMatch="ACCEPT"
onMismatch="DENY"/>
+ <PatternLayout pattern="${log_pattern}"/>
+ </Console>
+
+ <RollingFile name="DebugFile" fileName="${debug_fileName}"
filePattern="${debug_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${debug_max}" />
+ <Filters>
+ <ThresholdFilter level="WARN" onMatch="DENY"
onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="INFO" onMatch="DENY"
onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="DEBUG" onMatch="ACCEPT"
onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="InfoFile" fileName="${info_fileName}"
filePattern="${info_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${info_max}" />
+ <Filters>
+ <ThresholdFilter level="WARN" onMatch="DENY"
onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="INFO" onMatch="ACCEPT"
onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="WarnFile" fileName="${warn_fileName}"
filePattern="${warn_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${warn_max}" />
+ <Filters>
+ <ThresholdFilter level="ERROR" onMatch="DENY"
onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="WARN" onMatch="ACCEPT"
onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="ErrorFile" fileName="${error_fileName}"
filePattern="${error_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${error_max}" />
+ <Filters>
+ <ThresholdFilter level="FATAL" onMatch="DENY"
onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="ERROR" onMatch="ACCEPT"
onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="PutCounterGroup"
fileName="${put_transfer_fileName}" filePattern="${put_transfer_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${put_transfer_max}" />
+ <Filters>
+ <ThresholdFilter level="WARN" onMatch="DENY"
onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="INFO" onMatch="ACCEPT"
onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="GetCounterGroup"
fileName="${get_transfer_fileName}" filePattern="${get_transfer_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${get_transfer_max}" />
+ <Filters>
+ <ThresholdFilter level="WARN" onMatch="DENY"
onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="INFO" onMatch="ACCEPT"
onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+ </appenders>
+
+ <loggers>
+ <root level="${output_log_level}">
+ <appender-ref ref="Console"/>
+ <appender-ref ref="DebugFile"/>
+ <appender-ref ref="InfoFile"/>
+ <appender-ref ref="WarnFile"/>
+ <appender-ref ref="ErrorFile"/>
+ </root>
+ </loggers>
+</configuration>
\ No newline at end of file
diff --git a/inlong-tubemq/conf/master.log4j.properties
b/inlong-tubemq/conf/master.log4j.properties
deleted file mode 100644
index 0f21c5b3d..000000000
--- a/inlong-tubemq/conf/master.log4j.properties
+++ /dev/null
@@ -1,25 +0,0 @@
-#
-# 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.
-#
-
-log4j.rootLogger=INFO, MasterDailyRollingFile
-log4j.appender.MasterDailyRollingFile=org.apache.log4j.DailyRollingFileAppender
-log4j.appender.MasterDailyRollingFile.DatePattern='.'yyyy-MM-dd
-log4j.appender.MasterDailyRollingFile.File=${tubemq.home}/logs/master.log
-log4j.appender.MasterDailyRollingFile.layout=org.apache.log4j.PatternLayout
-log4j.appender.MasterDailyRollingFile.layout.ConversionPattern=[%p]
%d{yyyy-MM-dd HH:mm:ss,SSS} [%c{1}] %m%n
-log4j.appender.MasterDailyRollingFile.Append=true
-
diff --git a/inlong-tubemq/conf/tools.log4j.properties
b/inlong-tubemq/conf/tools.log4j.properties
deleted file mode 100644
index bb49f2264..000000000
--- a/inlong-tubemq/conf/tools.log4j.properties
+++ /dev/null
@@ -1,24 +0,0 @@
-#
-# 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.
-#
-
-log4j.rootLogger=INFO, stdout
-log4j.appender.stdout=org.apache.log4j.ConsoleAppender
-log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
-
-
-
diff --git a/inlong-tubemq/pom.xml b/inlong-tubemq/pom.xml
index a953e8eb1..99831789d 100644
--- a/inlong-tubemq/pom.xml
+++ b/inlong-tubemq/pom.xml
@@ -72,19 +72,9 @@
</modules>
<properties>
- <junit.version>4.13.2</junit.version>
- <mockito.version>2.18.0</mockito.version>
- <powermock.version>2.0.4</powermock.version>
<exec.maven.version>1.6.0</exec.maven.version>
<plugin.assembly.version>3.3.0</plugin.assembly.version>
<build.helper.maven.version>3.0.0</build.helper.maven.version>
- <protobuf.version>3.19.4</protobuf.version>
- <je.version>7.3.7</je.version>
- <gson.version>2.8.5</gson.version>
- <slf4j.version>1.7.36</slf4j.version>
- <spring.version>5.3.18</spring.version>
- <spring.boot.version>2.6.6</spring.boot.version>
- <netty.version>3.10.6.Final</netty.version>
</properties>
<repositories>
@@ -168,214 +158,4 @@
</plugin>
</plugins>
</build>
- <dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>org.apache.inlong</groupId>
- <artifactId>tubemq-core</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.inlong</groupId>
- <artifactId>tubemq-client</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.inlong</groupId>
- <artifactId>tubemq-example</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- <version>${slf4j.version}</version>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- <version>${slf4j.version}</version>
- </dependency>
- <dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty</artifactId>
- <version>${netty.version}</version>
- </dependency>
- <dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- <version>${protobuf.version}</version>
- </dependency>
- <dependency>
- <groupId>commons-cli</groupId>
- <artifactId>commons-cli</artifactId>
- <version>1.4</version>
- </dependency>
- <dependency>
- <groupId>commons-codec</groupId>
- <artifactId>commons-codec</artifactId>
- <version>1.10</version>
- </dependency>
- <dependency>
- <groupId>commons-lang</groupId>
- <artifactId>commons-lang</artifactId>
- <version>2.6</version>
- </dependency>
- <dependency>
- <groupId>com.sleepycat</groupId>
- <artifactId>je</artifactId>
- <version>${je.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>com.google.code.gson</groupId>
- <artifactId>gson</artifactId>
- <version>${gson.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
- <version>4.5.13</version>
- </dependency>
- <dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- <version>2.11.0</version>
- </dependency>
- <dependency>
- <groupId>org.apache.zookeeper</groupId>
- <artifactId>zookeeper</artifactId>
- <version>3.4.14</version>
- <exclusions>
- <exclusion>
- <groupId>com.google.code.findbugs</groupId>
- <artifactId>jsr305</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.github.spotbugs</groupId>
- <artifactId>spotbugs-annotations</artifactId>
- </exclusion>
- <exclusion>
- <groupId>jline</groupId>
- <artifactId>jline</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.sun.jmx</groupId>
- <artifactId>jmxri</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.sun.jdmk</groupId>
- <artifactId>jmxtools</artifactId>
- </exclusion>
- <exclusion>
- <groupId>javax.jms</groupId>
- <artifactId>jms</artifactId>
- </exclusion>
- <exclusion>
- <groupId>io.netty</groupId>
- <artifactId>netty</artifactId>
- </exclusion>
- <exclusion>
- <groupId>io.netty</groupId>
- <artifactId>netty-all</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>13.0</version>
- </dependency>
- <dependency>
- <groupId>javax.servlet</groupId>
- <artifactId>javax.servlet-api</artifactId>
- <version>4.0.1</version>
- </dependency>
- <dependency>
- <groupId>org.apache.velocity.tools</groupId>
- <artifactId>velocity-tools-generic</artifactId>
- <version>3.1</version>
- </dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-core</artifactId>
- <version>${spring.version}</version>
- </dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-context</artifactId>
- <version>${spring.version}</version>
- </dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-jdbc</artifactId>
- <version>${spring.version}</version>
- </dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-orm</artifactId>
- <version>${spring.version}</version>
- </dependency>
- <dependency>
- <groupId>org.mortbay.jetty</groupId>
- <artifactId>jetty</artifactId>
- <version>6.1.26</version>
- <exclusions>
- <exclusion>
- <groupId>org.mortbay.jetty</groupId>
- <artifactId>jsp-2.1</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.mortbay.jetty</groupId>
- <artifactId>jsp-api-2.1</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.mortbay.jetty</groupId>
- <artifactId>servlet-api</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.velocity</groupId>
- <artifactId>velocity-engine-core</artifactId>
- <version>2.3</version>
- </dependency>
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-server</artifactId>
- <version>9.4.44.v20210927</version>
- </dependency>
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-servlet</artifactId>
- <version>9.4.44.v20210927</version>
- </dependency>
- <dependency>
- <groupId>org.ini4j</groupId>
- <artifactId>ini4j</artifactId>
- <version>0.5.1</version>
- </dependency>
- <dependency>
- <groupId>org.dom4j</groupId>
- <artifactId>dom4j</artifactId>
- <version>2.1.3</version>
- </dependency>
- <dependency>
- <groupId>org.easymock</groupId>
- <artifactId>easymock</artifactId>
- <version>2.5.2</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.easymock</groupId>
- <artifactId>easymockclassextension</artifactId>
- <version>2.5.2</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
</project>
diff --git a/inlong-tubemq/tubemq-client/pom.xml
b/inlong-tubemq/tubemq-client/pom.xml
index 0200a27fb..d4463d9ed 100644
--- a/inlong-tubemq/tubemq-client/pom.xml
+++ b/inlong-tubemq/tubemq-client/pom.xml
@@ -101,31 +101,32 @@
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-core</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
- <version>${junit.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
- <version>${mockito.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-junit4</artifactId>
- <version>${powermock.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-api-mockito2</artifactId>
- <version>${powermock.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-common</artifactId>
+ </dependency>
</dependencies>
<profiles>
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java
index 34619ee6f..68cf861eb 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java
@@ -17,6 +17,10 @@
package org.apache.inlong.tubemq.client.consumer;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.Timer;
+import io.netty.util.TimerTask;
import java.io.Closeable;
import java.util.ArrayList;
import java.util.HashMap;
@@ -46,10 +50,6 @@ import
org.apache.inlong.tubemq.corebase.utils.DataConverterUtil;
import org.apache.inlong.tubemq.corebase.utils.TStringUtils;
import org.apache.inlong.tubemq.corebase.utils.ThreadUtils;
import org.apache.inlong.tubemq.corebase.utils.Tuple2;
-import org.jboss.netty.util.HashedWheelTimer;
-import org.jboss.netty.util.Timeout;
-import org.jboss.netty.util.Timer;
-import org.jboss.netty.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/inlong-tubemq/tubemq-connectors/pom.xml
b/inlong-tubemq/tubemq-connectors/pom.xml
index af6bba759..1a0add6a7 100644
--- a/inlong-tubemq/tubemq-connectors/pom.xml
+++ b/inlong-tubemq/tubemq-connectors/pom.xml
@@ -39,11 +39,13 @@
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-client</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-core</artifactId>
+ <version>${project.version}</version>
</dependency>
</dependencies>
diff --git a/inlong-tubemq/tubemq-connectors/tubemq-connector-flink/pom.xml
b/inlong-tubemq/tubemq-connectors/tubemq-connector-flink/pom.xml
index df1ee5145..bb941cfd1 100644
--- a/inlong-tubemq/tubemq-connectors/tubemq-connector-flink/pom.xml
+++ b/inlong-tubemq/tubemq-connectors/tubemq-connector-flink/pom.xml
@@ -37,17 +37,18 @@
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-client</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-core</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
- <version>${flink.version}</version>
<scope>provided</scope>
</dependency>
diff --git a/inlong-tubemq/tubemq-connectors/tubemq-connector-spark/pom.xml
b/inlong-tubemq/tubemq-connectors/tubemq-connector-spark/pom.xml
index 077e19802..16861bc0c 100644
--- a/inlong-tubemq/tubemq-connectors/tubemq-connector-spark/pom.xml
+++ b/inlong-tubemq/tubemq-connectors/tubemq-connector-spark/pom.xml
@@ -37,11 +37,13 @@
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-client</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-core</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
@@ -65,7 +67,6 @@
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
- <version>${junit.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
diff --git a/inlong-tubemq/tubemq-core/pom.xml
b/inlong-tubemq/tubemq-core/pom.xml
index 93123da87..e776b9f1f 100644
--- a/inlong-tubemq/tubemq-core/pom.xml
+++ b/inlong-tubemq/tubemq-core/pom.xml
@@ -93,21 +93,25 @@
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
</dependency>
- <dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty</artifactId>
- </dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
<dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
+ <artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
<groupId>commons-codec</groupId>
@@ -120,9 +124,24 @@
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
- <version>${junit.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-handler</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport-native-epoll</artifactId>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/AddressUtils.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/AddressUtils.java
index 970d7cdad..1b0b1e9aa 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/AddressUtils.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/AddressUtils.java
@@ -17,6 +17,7 @@
package org.apache.inlong.tubemq.corebase.utils;
+import io.netty.channel.Channel;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
@@ -25,7 +26,6 @@ import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Enumeration;
import org.apache.inlong.tubemq.corebase.exception.AddressException;
-import org.jboss.netty.channel.Channel;
public class AddressUtils {
@@ -118,7 +118,7 @@ public class AddressUtils {
if (channel == null) {
return strRemoteIP;
}
- SocketAddress remoteSocketAddress = channel.getRemoteAddress();
+ SocketAddress remoteSocketAddress = channel.remoteAddress();
if (null != remoteSocketAddress) {
strRemoteIP = remoteSocketAddress.toString();
try {
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/RpcConstants.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/RpcConstants.java
index 80e451dfa..393863a0d 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/RpcConstants.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/RpcConstants.java
@@ -43,7 +43,9 @@ public final class RpcConstants {
public static final String NETTY_WRITE_HIGH_MARK =
"rpc.netty.write.highmark";
public static final String NETTY_WRITE_LOW_MARK =
"rpc.netty.write.lowmark";
public static final String NETTY_TCP_SENDBUF = "rpc.netty.send.buffer";
+ public static final String NETTY_TCP_MAX_MESSAGE_SIZE =
"rpc.netty.max.message.size";
public static final String NETTY_TCP_RECEIVEBUF =
"rpc.netty.receive.buffer";
+ public static final String NETTY_TCP_ENABLEBUSYWAIT =
"rpc.netty.enable.busy.wait";
public static final String TCP_NODELAY = "rpc.tcp.nodelay";
public static final String TCP_REUSEADDRESS = "rpc.tcp.reuseaddress";
@@ -131,5 +133,6 @@ public final class RpcConstants {
public static final long CFG_UNAVAILABLE_FORBIDDEN_DURATION_MS = 50000;
public static final long CFG_DEFAULT_NETTY_WRITEBUFFER_HIGH_MARK = 50 *
1024 * 1024;
public static final long CFG_DEFAULT_NETTY_WRITEBUFFER_LOW_MARK = 5 * 1024
* 1024;
+ public static final int CFG_DEFAULT_NETTY_TCP_MAX_MESSAGE_SIZE = 5 * 1024
* 1024;
}
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/EventLoopUtil.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/EventLoopUtil.java
new file mode 100644
index 000000000..1e9490efc
--- /dev/null
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/EventLoopUtil.java
@@ -0,0 +1,115 @@
+/**
+ * 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.inlong.tubemq.corerpc.netty;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.Epoll;
+import io.netty.channel.epoll.EpollChannelOption;
+import io.netty.channel.epoll.EpollDatagramChannel;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollMode;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.epoll.EpollSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.DatagramChannel;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.concurrent.Future;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ThreadFactory;
+
+public class EventLoopUtil {
+ public EventLoopUtil() {
+ }
+
+ public static EventLoopGroup newEventLoopGroup(int nThreads,
+ boolean enableBusyWait, ThreadFactory threadFactory) {
+ if (!Epoll.isAvailable()) {
+ return new NioEventLoopGroup(nThreads, threadFactory);
+ } else if (!enableBusyWait) {
+ return new EpollEventLoopGroup(nThreads, threadFactory);
+ } else {
+ EpollEventLoopGroup eventLoopGroup = new
EpollEventLoopGroup(nThreads,
+ threadFactory, () -> {
+ return (selectSupplier, hasTasks) -> {
+ return -3;
+ };
+ });
+ return eventLoopGroup;
+ }
+ }
+
+ public static Class<? extends SocketChannel>
getClientSocketChannelClass(EventLoopGroup eventLoopGroup) {
+ return eventLoopGroup instanceof EpollEventLoopGroup
+ ? EpollSocketChannel.class : NioSocketChannel.class;
+ }
+
+ public static Class<? extends ServerSocketChannel>
getServerSocketChannelClass(EventLoopGroup eventLoopGroup) {
+ return eventLoopGroup instanceof EpollEventLoopGroup
+ ? EpollServerSocketChannel.class :
NioServerSocketChannel.class;
+ }
+
+ public static Class<? extends DatagramChannel>
getDatagramChannelClass(EventLoopGroup eventLoopGroup) {
+ return eventLoopGroup instanceof EpollEventLoopGroup
+ ? EpollDatagramChannel.class : NioDatagramChannel.class;
+ }
+
+ public static void enableTriggeredMode(ServerBootstrap bootstrap) {
+ if (Epoll.isAvailable()) {
+ bootstrap.childOption(EpollChannelOption.EPOLL_MODE,
EpollMode.LEVEL_TRIGGERED);
+ }
+
+ }
+
+ public static CompletableFuture<Void> shutdownGracefully(EventLoopGroup
eventLoopGroup) {
+ return toCompletableFutureVoid(eventLoopGroup.shutdownGracefully());
+ }
+
+ /**
+ * get CompletableFuture by Future
+ *
+ * @param future Future
+ * @return CompletableFuture
+ */
+ public static CompletableFuture<Void> toCompletableFutureVoid(Future<?>
future) {
+ Objects.requireNonNull(future, "future cannot be null");
+
+ CompletableFuture<Void> adapter = new CompletableFuture<>();
+ if (future.isDone()) {
+ if (future.isSuccess()) {
+ adapter.complete(null);
+ } else {
+ adapter.completeExceptionally(future.cause());
+ }
+ } else {
+ future.addListener(f -> {
+ if (f.isSuccess()) {
+ adapter.complete(null);
+ } else {
+ adapter.completeExceptionally(f.cause());
+ }
+ });
+ }
+ return adapter;
+ }
+}
\ No newline at end of file
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClient.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClient.java
index 6a144461d..0faff9a05 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClient.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClient.java
@@ -18,6 +18,14 @@
package org.apache.inlong.tubemq.corerpc.netty;
import com.google.protobuf.ByteString;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.ReadTimeoutException;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.Timer;
+import io.netty.util.TimerTask;
import java.io.EOFException;
import java.io.IOException;
import java.nio.channels.UnresolvedAddressException;
@@ -40,18 +48,6 @@ import org.apache.inlong.tubemq.corerpc.codec.PbEnDecoder;
import org.apache.inlong.tubemq.corerpc.exception.ClientClosedException;
import org.apache.inlong.tubemq.corerpc.exception.NetworkException;
import org.apache.inlong.tubemq.corerpc.utils.MixUtils;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelHandler;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.channel.ChannelStateEvent;
-import org.jboss.netty.channel.ExceptionEvent;
-import org.jboss.netty.channel.MessageEvent;
-import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
-import org.jboss.netty.handler.timeout.ReadTimeoutException;
-import org.jboss.netty.util.HashedWheelTimer;
-import org.jboss.netty.util.Timeout;
-import org.jboss.netty.util.Timer;
-import org.jboss.netty.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -148,7 +144,7 @@ public class NettyClient implements Client {
requests.put(request.getSerialNo(), future);
if (callback == null) {
try {
- getChannel().write(pack);
+ getChannel().writeAndFlush(pack);
return future.get(timeout, timeUnit);
} catch (Throwable e) {
Callback<ResponseWrapper> callback1 =
@@ -170,7 +166,7 @@ public class NettyClient implements Client {
timer.newTimeout(new
TimeoutTask(request.getSerialNo()), timeout, timeUnit));
inserted = true;
//write data after build Timeout to avoid one request
processed twice
- getChannel().write(pack);
+ getChannel().writeAndFlush(pack);
} catch (Throwable e) {
Callback<ResponseWrapper> callback1 =
requests.remove(request.getSerialNo());
@@ -209,8 +205,8 @@ public class NettyClient implements Client {
return (!this.closed.get()
&& channel != null
&& channel.isOpen()
- && channel.isBound()
- && channel.isConnected());
+ && channel.isWritable()
+ && channel.isActive());
}
@Override
@@ -279,7 +275,7 @@ public class NettyClient implements Client {
/**
* tube NettyClientHandler
*/
- public class NettyClientHandler extends SimpleChannelUpstreamHandler {
+ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
/**
* Invoked when a message object was received from a remote peer.
@@ -288,9 +284,11 @@ public class NettyClient implements Client {
* @param e the message event
*/
@Override
- public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
- if (e.getMessage() instanceof RpcDataPack) {
- RpcDataPack dataPack = (RpcDataPack) e.getMessage();
+ public void channelRead(ChannelHandlerContext ctx, Object e) {
+ logger.debug("client message receive!");
+ if (e instanceof RpcDataPack) {
+ logger.debug("RpcDataPack client message receive!");
+ RpcDataPack dataPack = (RpcDataPack) e;
Callback callback = requests.remove(dataPack.getSerialNo());
if (callback != null) {
Timeout timeout = timeouts.remove(dataPack.getSerialNo());
@@ -374,18 +372,18 @@ public class NettyClient implements Client {
}
/**
- * Invoked when an exception was raised by an I/O thread or a {@link
ChannelHandler}.
+ * Invoked when an exception was raised by an I/O thread
*
* @param ctx the channel handler context
* @param e the exception object
*/
@Override
- public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent
e) throws Exception {
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e)
throws Exception {
Throwable t = e.getCause();
if ((t instanceof IOException || t instanceof ReadTimeoutException
|| t instanceof UnresolvedAddressException)) {
if (t instanceof ReadTimeoutException) {
- logger.info("Close client {} due to idle.",
e.getChannel());
+ logger.info("Close client {} due to idle.", ctx.channel());
}
if (t instanceof UnresolvedAddressException) {
logger.info("UnresolvedAddressException for connect {}
closed.", addressInfo.getHostPortStr());
@@ -400,10 +398,9 @@ public class NettyClient implements Client {
* Invoked when a {@link Channel} was closed and all its related
resources were released.
*
* @param ctx the channel handler context
- * @param e the channel state event
*/
@Override
- public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent
e) throws Exception {
+ public void channelInactive(ChannelHandlerContext ctx) {
NettyClient.this.close();
}
}
@@ -427,12 +424,8 @@ public class NettyClient implements Client {
}
final Callback callback = requests.remove(serialNo);
if (callback != null) {
- channel.getPipeline().execute(new Runnable() {
- @Override
- public void run() {
- callback.handleError(new TimeoutException("Request is
timeout!"));
- }
- });
+ channel.eventLoop().execute(
+ () -> callback.handleError(new
TimeoutException("Request is timeout!")));
}
}
}
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClientFactory.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClientFactory.java
index 92044193e..941865067 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClientFactory.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClientFactory.java
@@ -17,6 +17,17 @@
package org.apache.inlong.tubemq.corerpc.netty;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.concurrent.DefaultThreadFactory;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
@@ -32,20 +43,6 @@ import org.apache.inlong.tubemq.corerpc.client.Client;
import org.apache.inlong.tubemq.corerpc.client.ClientFactory;
import org.apache.inlong.tubemq.corerpc.exception.LocalConnException;
import org.apache.inlong.tubemq.corerpc.utils.TSSLEngineUtil;
-import org.jboss.netty.bootstrap.ClientBootstrap;
-import org.jboss.netty.channel.ChannelFactory;
-import org.jboss.netty.channel.ChannelFuture;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.jboss.netty.channel.Channels;
-import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
-import org.jboss.netty.channel.socket.nio.NioWorkerPool;
-import org.jboss.netty.handler.execution.ExecutionHandler;
-import org.jboss.netty.handler.execution.MemoryAwareThreadPoolExecutor;
-import org.jboss.netty.handler.ssl.SslHandler;
-import org.jboss.netty.util.HashedWheelTimer;
-import org.jboss.netty.util.ThreadNameDeterminer;
-import org.jboss.netty.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,15 +57,12 @@ public class NettyClientFactory implements ClientFactory {
protected final ConcurrentHashMap<String, Client> clients =
new ConcurrentHashMap<>();
protected AtomicBoolean shutdown = new AtomicBoolean(true);
- private Timer timer = new HashedWheelTimer();
- private volatile AtomicBoolean init = new AtomicBoolean(true);
- private ChannelFactory channelFactory;
- private MemoryAwareThreadPoolExecutor eventExecutor;
+ private EventLoopGroup eventLoopGroup;
private ExecutorService bossExecutorService;
private ExecutorService workerExecutorService;
private AtomicInteger workerIdCounter = new AtomicInteger(0);
- private RpcConfig factoryConf;
// TSL encryption and need Two Way Authentic
+ private int maxMessageSize;
private boolean enableTLS = false;
private boolean needTwoWayAuthentic = false;
private String keyStorePath;
@@ -87,13 +81,11 @@ public class NettyClientFactory implements ClientFactory {
* @throws IllegalArgumentException the exception while configuring object
*/
public void configure(final RpcConfig conf) throws
IllegalArgumentException {
- if (this.init.compareAndSet(false, true)) {
- this.timer = new HashedWheelTimer();
- }
if (this.shutdown.compareAndSet(true, false)) {
- this.factoryConf = conf;
enableTLS = conf.getBoolean(RpcConstants.TLS_OVER_TCP, false);
needTwoWayAuthentic =
conf.getBoolean(RpcConstants.TLS_TWO_WAY_AUTHENTIC, false);
+ this.maxMessageSize =
conf.getInt(RpcConstants.NETTY_TCP_MAX_MESSAGE_SIZE,
+ RpcConstants.CFG_DEFAULT_NETTY_TCP_MAX_MESSAGE_SIZE);
if (enableTLS) {
trustStorePath =
conf.getString(RpcConstants.TLS_TRUSTSTORE_PATH);
trustStorePassword =
conf.getString(RpcConstants.TLS_TRUSTSTORE_PASSWORD);
@@ -120,23 +112,14 @@ public class NettyClientFactory implements ClientFactory {
conf.getInt(RpcConstants.CALLBACK_WORKER_COUNT, 3);
bossExecutorService = Executors.newCachedThreadPool();
workerExecutorService = Executors.newCachedThreadPool();
- this.channelFactory = new
NioClientSocketChannelFactory(bossExecutorService, bossCount,
- new NioWorkerPool(workerExecutorService, workerCount, new
ThreadNameDeterminer() {
- @Override
- public String determineThreadName(String
currentThreadName, String proposedThreadName)
- throws Exception {
- return new StringBuilder(256)
-
.append(conf.getString(RpcConstants.WORKER_THREAD_NAME,
-
RpcConstants.CFG_DEFAULT_WORKER_THREAD_NAME))
-
.append(workerIdCounter.incrementAndGet()).toString();
- }
- }));
- this.eventExecutor = new MemoryAwareThreadPoolExecutor(
- callbackCount,
- conf.getInt(RpcConstants.WORKER_MEM_SIZE,
- RpcConstants.CFG_DEFAULT_TOTAL_MEM_SIZE),
- conf.getInt(RpcConstants.WORKER_MEM_SIZE,
- RpcConstants.CFG_DEFAULT_TOTAL_MEM_SIZE));
+ String threadName = new StringBuilder(256)
+ .append(conf.getString(RpcConstants.WORKER_THREAD_NAME,
+ RpcConstants.CFG_DEFAULT_WORKER_THREAD_NAME))
+ .append(workerIdCounter.incrementAndGet()).toString();
+ eventLoopGroup = EventLoopUtil.newEventLoopGroup(workerCount,
+ conf.getBoolean(RpcConstants.NETTY_TCP_ENABLEBUSYWAIT,
false),
+ new DefaultThreadFactory(threadName,
+ Thread.currentThread().isDaemon()));
}
}
@@ -203,10 +186,6 @@ public class NettyClientFactory implements ClientFactory {
@Override
public void shutdown() {
- // stop timer
- if (this.init.compareAndSet(true, false)) {
- timer.stop();
- }
// shutdown and release network resources
if (this.shutdown.compareAndSet(false, true)) {
try {
@@ -226,12 +205,8 @@ public class NettyClientFactory implements ClientFactory {
if (this.workerExecutorService != null) {
this.workerExecutorService.shutdown();
}
- if (this.eventExecutor != null) {
- this.eventExecutor.shutdown();
- }
- } finally {
- this.channelFactory.releaseExternalResources();
- this.channelFactory.shutdown();
+ } catch (Exception e) {
+ logger.error("has exception ", e);
}
}
}
@@ -249,25 +224,27 @@ public class NettyClientFactory implements ClientFactory {
int connectTimeout, final RpcConfig conf)
throws Exception {
final NettyClient client =
new NettyClient(this, connectTimeout);
- ClientBootstrap clientBootstrap = new ClientBootstrap();
- clientBootstrap.setOption("tcpNoDelay", true);
- clientBootstrap.setOption("reuseAddress", true);
- clientBootstrap.setOption("connectTimeoutMillis", connectTimeout);
- clientBootstrap.setFactory(this.channelFactory);
- long nettyWriteHighMark =
- conf.getLong(RpcConstants.NETTY_WRITE_HIGH_MARK, -1);
- long nettyWriteLowMark =
- conf.getLong(RpcConstants.NETTY_WRITE_LOW_MARK, -1);
+ Bootstrap clientBootstrap = new Bootstrap();
+ clientBootstrap.group(eventLoopGroup);
+
clientBootstrap.channel(EventLoopUtil.getClientSocketChannelClass(eventLoopGroup));
+ clientBootstrap.option(ChannelOption.TCP_NODELAY, true);
+ clientBootstrap.option(ChannelOption.SO_REUSEADDR, true);
+ clientBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,
connectTimeout);
+
+ int nettyWriteHighMark =
+ conf.getInt(RpcConstants.NETTY_WRITE_HIGH_MARK, -1);
+ int nettyWriteLowMark =
+ conf.getInt(RpcConstants.NETTY_WRITE_LOW_MARK, -1);
if (nettyWriteHighMark > 0) {
- clientBootstrap.setOption("writeBufferHighWaterMark",
nettyWriteHighMark);
+ clientBootstrap.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK,
nettyWriteHighMark);
}
if (nettyWriteLowMark > 0) {
- clientBootstrap.setOption("writeBufferLowWaterMark",
nettyWriteLowMark);
+ clientBootstrap.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK,
nettyWriteLowMark);
}
- clientBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
+ clientBootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
- public ChannelPipeline getPipeline() throws Exception {
- ChannelPipeline pipeline = Channels.pipeline();
+ public void initChannel(SocketChannel socketChannel)throws
Exception {
+ ChannelPipeline pipeline = socketChannel.pipeline();
if (enableTLS) {
try {
SSLEngine sslEngine =
@@ -281,26 +258,26 @@ public class NettyClientFactory implements ClientFactory {
throw new Exception(t);
}
}
+ socketChannel.pipeline().addLast("frameDecoder", new
LengthFieldBasedFrameDecoder(maxMessageSize,
+ 0, 4, 0, 4));
+
// Encode the data
pipeline.addLast("protocolEncoder", new
NettyProtocolEncoder());
// Decode the bytes into a Rpc Data Pack
pipeline.addLast("protocolDecoder", new
NettyProtocolDecoder());
// handle the time out requests
- pipeline.addLast("readTimeoutHandler", new
ReadTimeoutHandler(timer,
+ pipeline.addLast("readTimeoutHandler", new ReadTimeoutHandler(
conf.getLong(RpcConstants.CONNECT_READ_IDLE_DURATION,
RpcConstants.CFG_CONNECT_READ_IDLE_TIME),
TimeUnit.MILLISECONDS));
- // execution handler
- pipeline.addLast("execution", new
ExecutionHandler(eventExecutor));
// tube netty client handler
pipeline.addLast("clientHandler", client.new
NettyClientHandler());
- return pipeline;
}
});
ChannelFuture future =
clientBootstrap.connect(new
InetSocketAddress(addressInfo.getHost(), addressInfo.getPort()));
future.awaitUninterruptibly(connectTimeout);
if (!future.isDone()) {
- future.cancel();
+ future.cancel(false);
throw new LocalConnException(new StringBuilder(256).append("Create
connection to ")
.append(addressInfo.getHostPortStr()).append("
timeout!").toString());
}
@@ -310,9 +287,10 @@ public class NettyClientFactory implements ClientFactory {
}
if (!future.isSuccess()) {
throw new LocalConnException(new StringBuilder(256).append("Create
connection to ")
- .append(addressInfo.getHostPortStr()).append("
error").toString(), future.getCause());
+ .append(addressInfo.getHostPortStr()).append("
error").toString(),
+ future.cause());
}
- client.setChannel(future.getChannel(), addressInfo);
+ client.setChannel(future.channel(), addressInfo);
return client;
}
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolDecoder.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolDecoder.java
index ae172f8ce..238677076 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolDecoder.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolDecoder.java
@@ -18,72 +18,70 @@
package org.apache.inlong.tubemq.corerpc.netty;
import static
org.apache.inlong.tubemq.corebase.utils.AddressUtils.getRemoteAddressIP;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageDecoder;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.inlong.tubemq.corerpc.RpcConstants;
import org.apache.inlong.tubemq.corerpc.RpcDataPack;
import org.apache.inlong.tubemq.corerpc.exception.UnknownProtocolException;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.handler.codec.frame.FrameDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class NettyProtocolDecoder extends FrameDecoder {
- private static final Logger logger =
- LoggerFactory.getLogger(NettyProtocolDecoder.class);
+public class NettyProtocolDecoder extends MessageToMessageDecoder<ByteBuf> {
+ private static final Logger logger =
LoggerFactory.getLogger(NettyProtocolDecoder.class);
+
private static final ConcurrentHashMap<String, AtomicLong>
errProtolAddrMap =
new ConcurrentHashMap<>();
private static final ConcurrentHashMap<String, AtomicLong> errSizeAddrMap =
new ConcurrentHashMap<>();
private static AtomicLong lastProtolTime = new AtomicLong(0);
private static AtomicLong lastSizeTime = new AtomicLong(0);
- private boolean packHeaderRead = false;
- private int listSize;
- private RpcDataPack dataPack;
@Override
- protected Object decode(ChannelHandlerContext ctx, Channel channel,
- ChannelBuffer buffer) throws Exception {
- if (!packHeaderRead) {
- if (buffer.readableBytes() < 12) {
- return null;
- }
- int frameToken = buffer.readInt();
- filterIllegalPkgToken(frameToken,
- RpcConstants.RPC_PROTOCOL_BEGIN_TOKEN, channel);
- int serialNo = buffer.readInt();
- int tmpListSize = buffer.readInt();
- filterIllegalPackageSize(true, tmpListSize,
- RpcConstants.MAX_FRAME_MAX_LIST_SIZE, channel);
- this.listSize = tmpListSize;
- this.dataPack = new RpcDataPack(serialNo, new
ArrayList<ByteBuffer>(this.listSize));
- this.packHeaderRead = true;
+ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer,
List<Object> out) throws Exception {
+ if (buffer.readableBytes() < 12) {
+ logger.warn("Decode buffer.readableBytes() < 12 !");
+ return;
}
+ int frameToken = buffer.readInt();
+ filterIllegalPkgToken(frameToken,
+ RpcConstants.RPC_PROTOCOL_BEGIN_TOKEN, ctx.channel());
+ int serialNo = buffer.readInt();
+ int tmpListSize = buffer.readInt();
+ filterIllegalPackageSize(true, tmpListSize,
+ RpcConstants.MAX_FRAME_MAX_LIST_SIZE, ctx.channel());
+ RpcDataPack dataPack = new RpcDataPack(serialNo, new
ArrayList<ByteBuffer>());
// get PackBody
- if (buffer.readableBytes() < 4) {
- return null;
- }
- buffer.markReaderIndex();
- int length = buffer.readInt();
- filterIllegalPackageSize(false, length,
- RpcConstants.RPC_MAX_BUFFER_SIZE, channel);
- if (buffer.readableBytes() < length) {
- buffer.resetReaderIndex();
- return null;
+ int i = 0;
+ while (i < tmpListSize) {
+ i++;
+ if (buffer.readableBytes() < 4) {
+ logger.warn("Decode buffer.readableBytes() < 4 !");
+ break;
+ }
+ buffer.markReaderIndex();
+ int length = buffer.readInt();
+ filterIllegalPackageSize(false, length,
+ RpcConstants.RPC_MAX_BUFFER_SIZE, ctx.channel());
+ ByteBuffer bb = ByteBuffer.allocate(length);
+ buffer.readBytes(bb);
+ bb.flip();
+ dataPack.getDataLst().add(bb);
}
- ByteBuffer bb = ByteBuffer.allocate(length);
- buffer.readBytes(bb);
- bb.flip();
- dataPack.getDataLst().add(bb);
- if (dataPack.getDataLst().size() == listSize) {
- packHeaderRead = false;
- return dataPack;
+
+ if (dataPack.getDataLst().size() == tmpListSize) {
+ out.add(dataPack);
} else {
- return null;
+ logger.warn("Decode dataPack.getDataLst().size()[{}] !=
tmpListSize [{}] !",
+ dataPack.getDataLst().size(), tmpListSize);
+ return;
}
}
@@ -112,7 +110,7 @@ public class NettyProtocolDecoder extends FrameDecoder {
}
throw new UnknownProtocolException(new StringBuilder(256)
.append("Unknown protocol exception for message frame,
channel.address = ")
- .append(channel.getRemoteAddress().toString()).toString());
+ .append(channel.remoteAddress().toString()).toString());
}
}
@@ -141,7 +139,7 @@ public class NettyProtocolDecoder extends FrameDecoder {
}
StringBuilder sBuilder = new StringBuilder(256)
.append("Unknown protocol exception for message listSize!
channel.address = ")
- .append(channel.getRemoteAddress().toString());
+ .append(channel.remoteAddress().toString());
if (isFrameSize) {
sBuilder.append(", Max list size=").append(allowSize)
.append(", request's list size=").append(inParamValue);
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoder.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoder.java
index 0e056aa11..fde7b523a 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoder.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoder.java
@@ -17,30 +17,45 @@
package org.apache.inlong.tubemq.corerpc.netty;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import org.apache.inlong.tubemq.corerpc.RpcConstants;
import org.apache.inlong.tubemq.corerpc.RpcDataPack;
-import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-public class NettyProtocolEncoder extends OneToOneEncoder {
+public class NettyProtocolEncoder extends MessageToMessageEncoder<RpcDataPack>
{
+
+ private static final Logger logger =
LoggerFactory.getLogger(NettyProtocolEncoder.class);
@Override
- protected Object encode(ChannelHandlerContext ctx,
- Channel channel, Object msg) throws Exception {
- RpcDataPack dataPack = (RpcDataPack) msg;
+ protected void encode(ChannelHandlerContext chx, RpcDataPack msg,
List<Object> out) {
+ RpcDataPack dataPack = msg;
List<ByteBuffer> origs = dataPack.getDataLst();
- List<ByteBuffer> bbs = new ArrayList<>(origs.size() * 2 + 1);
- bbs.add(getPackHeader(dataPack));
- for (ByteBuffer b : origs) {
- bbs.add(getLengthHeader(b));
- bbs.add(b);
+ ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
+ try {
+ byteOut.write(getPackHeader(dataPack).array());
+ Iterator<ByteBuffer> iter = origs.iterator();
+ while (iter.hasNext()) {
+ ByteBuffer entry = iter.next();
+ byteOut.write(getLengthHeader(entry).array());
+ byteOut.write(entry.array());
+ }
+ byte[] body = byteOut.toByteArray();
+ ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(4 + body.length);
+ buf.writeInt(body.length);
+ buf.writeBytes(body);
+ out.add(buf);
+ } catch (IOException e) {
+ logger.error("encode has exception ", e);
}
- return ChannelBuffers.wrappedBuffer(bbs.toArray(new
ByteBuffer[bbs.size()]));
}
private ByteBuffer getPackHeader(RpcDataPack dataPack) {
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRequestContext.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRequestContext.java
index 086fac8ee..fdbe8b56c 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRequestContext.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRequestContext.java
@@ -18,6 +18,9 @@
package org.apache.inlong.tubemq.corerpc.netty;
import com.google.protobuf.ByteString;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.SocketAddress;
@@ -29,9 +32,6 @@ import org.apache.inlong.tubemq.corerpc.ResponseWrapper;
import org.apache.inlong.tubemq.corerpc.RpcDataPack;
import org.apache.inlong.tubemq.corerpc.codec.PbEnDecoder;
import org.apache.inlong.tubemq.corerpc.server.RequestContext;
-import org.jboss.netty.channel.ChannelFuture;
-import org.jboss.netty.channel.ChannelFutureListener;
-import org.jboss.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,7 +54,7 @@ public class NettyRequestContext implements RequestContext {
@Override
public SocketAddress getRemoteAddress() {
- return this.ctx.getChannel().getRemoteAddress();
+ return this.ctx.channel().remoteAddress();
}
@Override
@@ -70,7 +70,7 @@ public class NettyRequestContext implements RequestContext {
logger.debug(new StringBuilder(512)
.append("Timeout,so give up send response to
client.RequestId:")
.append(request.getSerialNo()).append(".client:")
- .append(ctx.getChannel().getRemoteAddress())
+ .append(ctx.channel().remoteAddress())
.append(",process time:")
.append(System.currentTimeMillis() - receiveTime)
.append(",timeout:").append(request.getTimeout()).toString());
@@ -78,26 +78,26 @@ public class NettyRequestContext implements RequestContext {
return;
}
dataPack = new RpcDataPack(response.getSerialNo(),
prepareResponse(response));
- ChannelFuture wf = ctx.getChannel().write(dataPack);
+ ChannelFuture wf = ctx.channel().writeAndFlush(dataPack);
wf.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws
Exception {
if (!future.isSuccess()) {
- Throwable exception = future.getCause();
+ Throwable exception = future.cause();
if (exception != null) {
if (logger.isDebugEnabled()) {
if
(IOException.class.isAssignableFrom(exception.getClass())) {
logger.debug(new StringBuilder(512)
.append("server write response error.")
.append("reason: ")
- .append(future.getChannel().toString())
+ .append(future.channel().toString())
.append(exception.toString()).toString());
} else {
logger.debug(new StringBuilder(512)
.append("server write response error.")
.append("reason: ")
- .append(future.getChannel().toString())
- .append(future.getCause()).toString());
+ .append(future.channel().toString())
+ .append(future.cause()).toString());
}
}
}
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java
index 830f8e5d0..7812499d3 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java
@@ -19,6 +19,17 @@ package org.apache.inlong.tubemq.corerpc.netty;
import static
org.apache.inlong.tubemq.corebase.utils.AddressUtils.getRemoteAddressIP;
import com.google.protobuf.Message;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -26,7 +37,6 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import javax.net.ssl.SSLEngine;
@@ -44,19 +54,6 @@ import
org.apache.inlong.tubemq.corerpc.server.RequestContext;
import org.apache.inlong.tubemq.corerpc.server.ServiceRpcServer;
import org.apache.inlong.tubemq.corerpc.utils.MixUtils;
import org.apache.inlong.tubemq.corerpc.utils.TSSLEngineUtil;
-import org.jboss.netty.bootstrap.ServerBootstrap;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelHandler;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.jboss.netty.channel.DefaultChannelPipeline;
-import org.jboss.netty.channel.ExceptionEvent;
-import org.jboss.netty.channel.MessageEvent;
-import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
-import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
-import org.jboss.netty.handler.ssl.SslHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -73,7 +70,9 @@ public class NettyRpcServer implements ServiceRpcServer {
private final ConcurrentHashMap<Integer, Protocol> protocols =
new ConcurrentHashMap<>();
private ServerBootstrap bootstrap;
- private NioServerSocketChannelFactory channelFactory = null;
+ private EventLoopGroup acceptorGroup;
+ private EventLoopGroup workerGroup;
+ private boolean enableBusyWait;
private AtomicBoolean started = new AtomicBoolean(false);
private int protocolType = RpcProtocol.RPC_PROTOCOL_TCP;
private boolean isOverTLS;
@@ -82,6 +81,7 @@ public class NettyRpcServer implements ServiceRpcServer {
private boolean needTwoWayAuthentic = false;
private String trustStorePath = "";
private String trustStorePassword = "";
+ private int maxMessageSize;
/**
* create a server with rpc config info
@@ -113,36 +113,45 @@ public class NettyRpcServer implements ServiceRpcServer {
}
}
}
+ this.enableBusyWait =
conf.getBoolean(RpcConstants.NETTY_TCP_ENABLEBUSYWAIT, false);
+ this.maxMessageSize =
conf.getInt(RpcConstants.NETTY_TCP_MAX_MESSAGE_SIZE,
+ RpcConstants.CFG_DEFAULT_NETTY_TCP_MAX_MESSAGE_SIZE);
int bossCount =
conf.getInt(RpcConstants.BOSS_COUNT,
RpcConstants.CFG_DEFAULT_BOSS_COUNT);
int workerCount =
conf.getInt(RpcConstants.WORKER_COUNT,
RpcConstants.CFG_DEFAULT_SERVER_WORKER_COUNT);
- this.bootstrap =
- new ServerBootstrap(new
NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
- bossCount, Executors.newCachedThreadPool(),
workerCount));
- bootstrap.setOption("tcpNoDelay",
+ this.acceptorGroup = EventLoopUtil.newEventLoopGroup(bossCount, false,
+ new DefaultThreadFactory("tcpSource-nettyBoss-threadGroup"));
+ this.workerGroup = EventLoopUtil
+ .newEventLoopGroup(workerCount, enableBusyWait,
+ new
DefaultThreadFactory("tcpSource-nettyWorker-threadGroup"));
+ this.bootstrap = new ServerBootstrap();
+
bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(workerGroup));
+ EventLoopUtil.enableTriggeredMode(bootstrap);
+ bootstrap.group(acceptorGroup, workerGroup);
+ bootstrap.childOption(ChannelOption.TCP_NODELAY,
conf.getBoolean(RpcConstants.TCP_NODELAY, true));
- bootstrap.setOption("reuseAddress",
+ bootstrap.childOption(ChannelOption.SO_REUSEADDR,
conf.getBoolean(RpcConstants.TCP_REUSEADDRESS, true));
- long nettyWriteHighMark =
- conf.getLong(RpcConstants.NETTY_WRITE_HIGH_MARK, -1);
+ int nettyWriteHighMark =
+ conf.getInt(RpcConstants.NETTY_WRITE_HIGH_MARK, -1);
if (nettyWriteHighMark > 0) {
- bootstrap.setOption("writeBufferHighWaterMark",
nettyWriteHighMark);
+ bootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK,
nettyWriteHighMark);
}
- long nettyWriteLowMark =
- conf.getLong(RpcConstants.NETTY_WRITE_LOW_MARK, -1);
+ int nettyWriteLowMark =
+ conf.getInt(RpcConstants.NETTY_WRITE_LOW_MARK, -1);
if (nettyWriteLowMark > 0) {
- bootstrap.setOption("writeBufferLowWaterMark", nettyWriteLowMark);
+ bootstrap.childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK,
nettyWriteLowMark);
}
- long nettySendBuf = conf.getLong(RpcConstants.NETTY_TCP_SENDBUF, -1);
+ int nettySendBuf = conf.getInt(RpcConstants.NETTY_TCP_SENDBUF, -1);
if (nettySendBuf > 0) {
- bootstrap.setOption("sendBufferSize", nettySendBuf);
+ bootstrap.childOption(ChannelOption.SO_SNDBUF, nettySendBuf);
}
- long nettyRecvBuf = conf.getLong(RpcConstants.NETTY_TCP_RECEIVEBUF,
-1);
+ int nettyRecvBuf = conf.getInt(RpcConstants.NETTY_TCP_RECEIVEBUF, -1);
if (nettyRecvBuf > 0) {
- bootstrap.setOption("receiveBufferSize", nettyRecvBuf);
+ bootstrap.childOption(ChannelOption.SO_RCVBUF, nettyRecvBuf);
}
}
@@ -151,32 +160,32 @@ public class NettyRpcServer implements ServiceRpcServer {
if (this.started.get()) {
return;
}
- bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
+ bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
- public ChannelPipeline getPipeline() throws Exception {
- ChannelPipeline pipeline = new DefaultChannelPipeline();
+ public void initChannel(SocketChannel socketChannel) {
if (isOverTLS) {
try {
SSLEngine sslEngine =
TSSLEngineUtil.createSSLEngine(keyStorePath,
trustStorePath,
keyStorePassword, trustStorePassword,
false, needTwoWayAuthentic);
- pipeline.addLast("ssl", new SslHandler(sslEngine));
+ socketChannel.pipeline().addLast("ssl", new
SslHandler(sslEngine));
} catch (Throwable t) {
logger.error(
"TLS NettyRpcServer init SSLEngine error,
system auto exit!", t);
System.exit(1);
}
}
+ socketChannel.pipeline().addLast("frameDecoder", new
LengthFieldBasedFrameDecoder(
+ maxMessageSize, 0, 4, 0, 4));
// Encode the data handler
- pipeline.addLast("protocolEncoder", new
NettyProtocolDecoder());
+ socketChannel.pipeline().addLast("protocolEncoder", new
NettyProtocolDecoder());
// Decode the bytes into a Rpc Data Pack
- pipeline.addLast("protocolDecoder", new
NettyProtocolEncoder());
+ socketChannel.pipeline().addLast("protocolDecoder", new
NettyProtocolEncoder());
// tube netty Server handler
- pipeline.addLast("serverHandler", new
NettyServerHandler(protocolType));
- return pipeline;
+ socketChannel.pipeline().addLast("serverHandler", new
NettyServerHandler(protocolType));
}
});
- bootstrap.bind(new InetSocketAddress(listenPort));
+ bootstrap.bind(new InetSocketAddress(listenPort)).sync();
this.started.set(true);
if (isOverTLS) {
logger.info(new StringBuilder(256)
@@ -233,7 +242,6 @@ public class NettyRpcServer implements ServiceRpcServer {
}
if (this.started.compareAndSet(true, false)) {
logger.info("Stopping RpcServer...");
- bootstrap.releaseExternalResources();
logger.info("RpcServer stop successfully.");
}
}
@@ -241,7 +249,7 @@ public class NettyRpcServer implements ServiceRpcServer {
/**
* Netty Server Handler
*/
- private class NettyServerHandler extends SimpleChannelUpstreamHandler {
+ private class NettyServerHandler extends ChannelInboundHandlerAdapter {
private int protocolType = RpcProtocol.RPC_PROTOCOL_TCP;
@@ -251,31 +259,32 @@ public class NettyRpcServer implements ServiceRpcServer {
/**
* Invoked when an exception was raised by an I/O thread or a
- * {@link ChannelHandler}.
*/
@Override
- public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent
e) throws Exception {
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) {
if (!(e.getCause() instanceof IOException)) {
- logger.error("catch some exception not IOException",
e.getCause());
+ logger.error("catch some exception not IOException {}", e);
}
+ ctx.fireExceptionCaught(e);
}
/**
- * Invoked when a message object (e.g: {@link ChannelBuffer}) was
received
+ * Invoked when a message object was received
* from a remote peer.
*/
@Override
- public void messageReceived(final ChannelHandlerContext ctx,
- MessageEvent e) throws Exception {
- if (!(e.getMessage() instanceof RpcDataPack)) {
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
+ logger.debug("server message receive!");
+ if (!(msg instanceof RpcDataPack)) {
return;
}
- RpcDataPack dataPack = (RpcDataPack) e.getMessage();
+ logger.debug("server RpcDataPack message receive!");
+ RpcDataPack dataPack = (RpcDataPack) msg;
RPCProtos.RpcConnHeader connHeader;
RPCProtos.RequestHeader requestHeader;
RPCProtos.RequestBody rpcRequestBody;
int rmtVersion = RpcProtocol.RPC_PROTOCOL_VERSION;
- Channel channel = ctx.getChannel();
+ Channel channel = ctx.channel();
if (channel == null) {
return;
}
@@ -321,7 +330,7 @@ public class NettyRpcServer implements ServiceRpcServer {
.append(e1.getMessage()).toString());
if (res != null) {
dataPack.setDataLst(res);
- channel.write(dataPack);
+ channel.writeAndFlush(dataPack);
}
return;
}
@@ -345,7 +354,7 @@ public class NettyRpcServer implements ServiceRpcServer {
.append(ee.getMessage()).toString());
if (res != null) {
dataPack.setDataLst(res);
- ctx.getChannel().write(dataPack);
+ ctx.channel().writeAndFlush(dataPack);
}
return;
}
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/ReadTimeoutHandler.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/ReadTimeoutHandler.java
deleted file mode 100644
index 70b35649f..000000000
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/ReadTimeoutHandler.java
+++ /dev/null
@@ -1,244 +0,0 @@
-/**
- * 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.inlong.tubemq.corerpc.netty;
-
-import static org.jboss.netty.channel.Channels.fireExceptionCaught;
-import java.util.concurrent.TimeUnit;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.channel.ChannelStateEvent;
-import org.jboss.netty.channel.LifeCycleAwareChannelHandler;
-import org.jboss.netty.channel.MessageEvent;
-import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
-import org.jboss.netty.handler.timeout.ReadTimeoutException;
-import org.jboss.netty.util.ExternalResourceReleasable;
-import org.jboss.netty.util.Timeout;
-import org.jboss.netty.util.Timer;
-import org.jboss.netty.util.TimerTask;
-
-public class ReadTimeoutHandler extends SimpleChannelUpstreamHandler implements
- LifeCycleAwareChannelHandler, ExternalResourceReleasable {
-
- private final Timer timer;
- private final long timeoutMillis;
-
- /**
- * Creates a new instance.
- *
- * @param timer the {@link Timer} that is used to trigger the
scheduled event. The
- * recommended {@link Timer} implementation is {@link
- * org.jboss.netty.util.HashedWheelTimer}.
- * @param timeoutSeconds read timeout in seconds
- */
- public ReadTimeoutHandler(Timer timer, int timeoutSeconds) {
- this(timer, timeoutSeconds, TimeUnit.SECONDS);
- }
-
- /**
- * Creates a new instance.
- *
- * @param timer the {@link Timer} that is used to trigger the scheduled
event. The recommended
- * {@link Timer} implementation is {@link
org.jboss.netty.util.HashedWheelTimer}.
- * @param timeout read timeout
- * @param unit the {@link TimeUnit} of {@code timeout}
- */
- public ReadTimeoutHandler(Timer timer, long timeout, TimeUnit unit) {
- if (timer == null) {
- throw new NullPointerException("timer");
- }
- if (unit == null) {
- throw new NullPointerException("unit");
- }
-
- this.timer = timer;
- if (timeout <= 0) {
- timeoutMillis = 0;
- } else {
- timeoutMillis = Math.max(unit.toMillis(timeout), 1);
- }
- }
-
- private static void destroy(ChannelHandlerContext ctx) {
- State state = state(ctx);
- synchronized (state) {
- if (state.state != 1) {
- return;
- }
- state.state = 2;
- }
-
- if (state.timeout != null) {
- state.timeout.cancel();
- state.timeout = null;
- }
- }
-
- private static State state(ChannelHandlerContext ctx) {
- State state;
- synchronized (ctx) {
- // TODO: It could have been better if there is
setAttachmentIfAbsent().
- state = (State) ctx.getAttachment();
- if (state != null) {
- return state;
- }
- state = new State();
- ctx.setAttachment(state);
- }
- return state;
- }
-
- /**
- * Stops the {@link Timer} which was specified in the constructor of this
handler. You should
- * not call this method if the {@link Timer} is in use by other objects.
- */
- @Override
- public void releaseExternalResources() {
- if (timer != null) {
- timer.stop();
- }
- }
-
- @Override
- public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
- if (ctx.getPipeline().isAttached()) {
- // channelOpen event has been fired already, which means
- // this.channelOpen() will not be invoked.
- // We have to initialize here instead.
- initialize(ctx);
- } else {
- // channelOpen event has not been fired yet.
- // this.channelOpen() will be invoked and initialization will
occur there.
- }
- }
-
- @Override
- public void afterAdd(ChannelHandlerContext ctx) throws Exception {
- // NOOP
- }
-
- @Override
- public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
- destroy(ctx);
- }
-
- @Override
- public void afterRemove(ChannelHandlerContext ctx) throws Exception {
- // NOOP
- }
-
- @Override
- public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
- // This method will be invoked only if this handler was added
- // before channelOpen event is fired. If a user adds this handler
- // after the channelOpen event, initialize() will be called by
beforeAdd().
- initialize(ctx);
- ctx.sendUpstream(e);
- }
-
- @Override
- public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
- destroy(ctx);
- ctx.sendUpstream(e);
- }
-
- @Override
- public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
- State state = (State) ctx.getAttachment();
- state.lastReadTime = System.currentTimeMillis();
- ctx.sendUpstream(e);
- }
-
- private void initialize(ChannelHandlerContext ctx) {
- State state = state(ctx);
-
- // Avoid the case where destroy() is called before scheduling timeouts.
- // See: https://github.com/netty/netty/issues/143
- synchronized (state) {
- switch (state.state) {
- case 1:
- case 2:
- return;
- }
- state.state = 1;
- }
-
- if (timeoutMillis > 0) {
- state.timeout =
- timer.newTimeout(new ReadTimeoutTask(ctx), timeoutMillis,
TimeUnit.MILLISECONDS);
- }
- }
-
- protected void readTimedOut(ChannelHandlerContext ctx) throws Exception {
- fireExceptionCaught(ctx, new ReadTimeoutException());
- }
-
- private static final class State {
- // 0 - none, 1 - initialized, 2 - destroyed
- int state;
- volatile Timeout timeout;
- volatile long lastReadTime = System.currentTimeMillis();
-
- State() {
- }
- }
-
- private final class ReadTimeoutTask implements TimerTask {
-
- private final ChannelHandlerContext ctx;
-
- ReadTimeoutTask(ChannelHandlerContext ctx) {
- this.ctx = ctx;
- }
-
- @Override
- public void run(Timeout timeout) throws Exception {
- if (timeout.isCancelled()) {
- return;
- }
-
- if (!ctx.getChannel().isOpen()) {
- return;
- }
-
- State state = (State) ctx.getAttachment();
- long currentTime = System.currentTimeMillis();
- long nextDelay = timeoutMillis - (currentTime -
state.lastReadTime);
- if (nextDelay <= 0) {
- // Read timed out - set a new timeout and notify the callback.
- state.timeout = timer.newTimeout(this, timeoutMillis,
TimeUnit.MILLISECONDS);
- fireReadTimedOut(ctx);
- } else {
- // Read occurred before the timeout - set a new timeout with
shorter delay.
- state.timeout = timer.newTimeout(this, nextDelay,
TimeUnit.MILLISECONDS);
- }
- }
-
- private void fireReadTimedOut(final ChannelHandlerContext ctx) throws
Exception {
- ctx.getPipeline().execute(new Runnable() {
-
- @Override
- public void run() {
- try {
- readTimedOut(ctx);
- } catch (Throwable t) {
- fireExceptionCaught(ctx, t);
- }
- }
- });
- }
- }
-}
diff --git
a/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoderTest.java
b/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoderTest.java
index 44e0c2f91..07b44a819 100644
---
a/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoderTest.java
+++
b/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoderTest.java
@@ -17,11 +17,12 @@
package org.apache.inlong.tubemq.corerpc.netty;
+import io.netty.buffer.ByteBuf;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import org.apache.inlong.tubemq.corerpc.RpcDataPack;
-import org.jboss.netty.buffer.ChannelBuffer;
import org.junit.Assert;
import org.junit.Test;
@@ -42,13 +43,15 @@ public class NettyProtocolEncoderTest {
dataList.add(ByteBuffer.wrap("def".getBytes()));
// append data list.
obj.setDataLst(dataList);
+ List<Object> out = new ArrayList<>();
try {
// encode data
- Object result = nettyProtocolEncoder.encode(null, null, obj);
- ChannelBuffer buf = (ChannelBuffer) result;
+ nettyProtocolEncoder.encode(null, obj, out);
+ ByteBuf buf = (ByteBuf) out.get(0);
// read data.
int i = buf.readInt();
i = buf.readInt();
+ i = buf.readInt();
Assert.assertEquals(123, i);
} catch (Exception e) {
e.printStackTrace();
diff --git a/inlong-tubemq/tubemq-example/pom.xml
b/inlong-tubemq/tubemq-example/pom.xml
index fb48ffc79..2dff6c27c 100644
--- a/inlong-tubemq/tubemq-example/pom.xml
+++ b/inlong-tubemq/tubemq-example/pom.xml
@@ -64,6 +64,7 @@
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-client</artifactId>
+ <version>${project.version}</version>
</dependency>
</dependencies>
diff --git a/inlong-tubemq/tubemq-server/pom.xml
b/inlong-tubemq/tubemq-server/pom.xml
index ddc346363..40acca82d 100644
--- a/inlong-tubemq/tubemq-server/pom.xml
+++ b/inlong-tubemq/tubemq-server/pom.xml
@@ -115,18 +115,59 @@
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-core</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-client</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-example</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.github.spotbugs</groupId>
+ <artifactId>spotbugs-annotations</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>jline</groupId>
+ <artifactId>jline</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.sun.jmx</groupId>
+ <artifactId>jmxri</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.sun.jdmk</groupId>
+ <artifactId>jmxtools</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>javax.jms</groupId>
+ <artifactId>jms</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>netty</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.ini4j</groupId>
@@ -156,6 +197,10 @@
<groupId>javax.servlet</groupId>
<artifactId>javax.servlet-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-jcl</artifactId>
+ </dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
@@ -199,13 +244,11 @@
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
- <version>${junit.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
- <version>${mockito.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
diff --git a/pom.xml b/pom.xml
index e7d0639f2..e774c0d29 100644
--- a/pom.xml
+++ b/pom.xml
@@ -117,6 +117,9 @@
<docker.organization>inlong</docker.organization>
<netty.version>4.1.72.Final</netty.version>
+ <scala.binary.version>2.11</scala.binary.version>
+ <spark.version>2.4.4</spark.version>
+
<simpleclient.httpserver.version>0.14.1</simpleclient.httpserver.version>
<httpcore.version>4.4.14</httpcore.version>
<httpclient.version>4.5.13</httpclient.version>
@@ -165,8 +168,9 @@
<mortbay.jetty.servlet.api.version>3.0.20100224</mortbay.jetty.servlet.api.version>
<mortbay.jetty.servlet.version>6.1.26</mortbay.jetty.servlet.version>
<jetty.version>9.4.44.v20210927</jetty.version>
+ <jetty.servlet.version>2.5-20110124</jetty.servlet.version>
<opencsv.version>5.4</opencsv.version>
- <javax.servlet.version>2.5</javax.servlet.version>
+ <javax.servlet.api.version>4.0.1</javax.servlet.api.version>
<gson.version>2.8.6</gson.version>
<jackson.version>2.13.1</jackson.version>
@@ -177,6 +181,13 @@
<hive.version>3.1.2</hive.version>
<flume.version>1.9.0</flume.version>
+ <velocity.tools.generic.version>3.1</velocity.tools.generic.version>
+ <velocity.engine.core.version>2.3</velocity.engine.core.version>
+
+ <ini4j.version>0.5.1</ini4j.version>
+ <dom4j.version>2.1.3</dom4j.version>
+
+ <zookeeper.version>3.6.3</zookeeper.version>
<pulsar.version>2.8.1</pulsar.version>
<pulsar.testcontainers.version>1.15.3</pulsar.testcontainers.version>
<kafka.version>2.4.1</kafka.version>
@@ -197,7 +208,7 @@
<antlr.verison>4.0.4</antlr.verison>
<logback.version>1.2.10</logback.version>
- <log4j2.version>2.17.1</log4j2.version>
+ <log4j2.version>2.17.2</log4j2.version>
<slf4j.version>1.7.36</slf4j.version>
<junit.version>4.13.2</junit.version>
@@ -211,6 +222,7 @@
<hamcrest.version>1.3</hamcrest.version>
<jsr.version>3.0.2</jsr.version>
<jcommander.version>1.78</jcommander.version>
+ <je.version>7.3.7</je.version>
</properties>
<dependencyManagement>
@@ -516,6 +528,26 @@
<version>${spring.version}</version>
<scope>compile</scope>
</dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-core</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-context</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-jdbc</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-orm</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-beans</artifactId>
@@ -628,9 +660,8 @@
<dependency>
<groupId>javax.servlet</groupId>
- <artifactId>servlet-api</artifactId>
- <version>${javax.servlet.version}</version>
- <scope>provided</scope>
+ <artifactId>javax.servlet-api</artifactId>
+ <version>${javax.servlet.api.version}</version>
</dependency>
<!-- jetty -->
@@ -655,7 +686,6 @@
<artifactId>opencsv</artifactId>
<version>${opencsv.version}</version>
</dependency>
-
<!-- http -->
<dependency>
<groupId>org.apache.httpcomponents</groupId>
@@ -695,6 +725,24 @@
<version>${elasticsearch.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-core_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+
<artifactId>spark-streaming_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ </dependency>
+
<!--pulsar dependency-->
<dependency>
<groupId>org.apache.pulsar</groupId>
@@ -926,6 +974,11 @@
</dependency>
<!-- zookeeper -->
+ <dependency>
+ <groupId>org.apache.zookeeper</groupId>
+ <artifactId>zookeeper</artifactId>
+ <version>${zookeeper.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
@@ -1007,6 +1060,43 @@
<artifactId>simpleclient_hotspot</artifactId>
<version>${simpleclient.httpserver.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.sleepycat</groupId>
+ <artifactId>je</artifactId>
+ <version>${je.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.velocity.tools</groupId>
+ <artifactId>velocity-tools-generic</artifactId>
+ <version>${velocity.tools.generic.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.velocity</groupId>
+ <artifactId>velocity-engine-core</artifactId>
+ <version>${velocity.engine.core.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-server</artifactId>
+ <version>${jetty.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-servlet</artifactId>
+ <version>${jetty.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.ini4j</groupId>
+ <artifactId>ini4j</artifactId>
+ <version>${ini4j.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.dom4j</groupId>
+ <artifactId>dom4j</artifactId>
+ <version>${dom4j.version}</version>
+ </dependency>
<!-- log 4j2 -->
<dependency>
@@ -1041,6 +1131,11 @@
<version>${log4j2.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-jcl</artifactId>
+ <version>${log4j2.version}</version>
+ </dependency>
<!-- logback -->
<dependency>
