This is an automated email from the ASF dual-hosted git repository.
dockerzhang 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 aa9f798ea [INLONG-3543][Audit] Upgrade netty and log4j version of
audit (#3545)
aa9f798ea is described below
commit aa9f798ea80c7aff48f162251f8a63d03374fbed
Author: baomingyu <[email protected]>
AuthorDate: Wed Apr 6 15:29:30 2022 +0800
[INLONG-3543][Audit] Upgrade netty and log4j version of audit (#3545)
---
inlong-audit/audit-common/pom.xml | 4 +
.../org/apache/inlong/audit/protocol/Commands.java | 17 ++-
inlong-audit/audit-proxy/pom.xml | 56 +++++----
.../org/apache/inlong/audit/sink/PulsarSink.java | 2 +-
.../inlong/audit/source/DefaultServiceDecoder.java | 6 +-
.../inlong/audit/source/ServerMessageFactory.java | 54 ++++-----
.../inlong/audit/source/ServerMessageHandler.java | 59 +++++-----
.../apache/inlong/audit/source/ServiceDecoder.java | 6 +-
.../inlong/audit/source/SimpleTcpSource.java | 111 +++++++++---------
.../apache/inlong/audit/utils/EventLoopUtil.java | 114 +++++++++++++++++++
inlong-audit/audit-sdk/pom.xml | 26 +++--
.../inlong/audit/send/ClientPipelineFactory.java | 39 +++++++
.../apache/inlong/audit/send/SenderChannel.java | 4 +-
.../org/apache/inlong/audit/send/SenderGroup.java | 83 +++++++-------
.../apache/inlong/audit/send/SenderHandler.java | 22 ++--
.../apache/inlong/audit/send/SenderManager.java | 34 +++---
.../java/org/apache/inlong/audit/util/Decoder.java | 25 ++--
.../java/org/apache/inlong/audit/util/Encoder.java | 39 -------
.../apache/inlong/audit/util/EventLoopUtil.java | 114 +++++++++++++++++++
.../java/org/apache/inlong/audit/util/IpPort.java | 8 +-
.../apache/inlong/audit/send/SenderGroupTest.java | 3 +-
inlong-audit/audit-store/pom.xml | 31 ++---
inlong-audit/bin/audit-proxy | 4 +-
inlong-audit/bin/store-start.sh | 5 +-
inlong-audit/conf/log4j2.xml | 126 +++++++++++++++++++++
inlong-audit/conf/logback.xml | 98 ----------------
inlong-audit/pom.xml | 77 ++++++++++++-
.../dataproxy/source/ServerMessageHandler.java | 16 +--
.../inlong/sdk/dataproxy/DefaultMessageSender.java | 1 -
29 files changed, 743 insertions(+), 441 deletions(-)
diff --git a/inlong-audit/audit-common/pom.xml
b/inlong-audit/audit-common/pom.xml
index 01cf90812..628ed7ab0 100644
--- a/inlong-audit/audit-common/pom.xml
+++ b/inlong-audit/audit-common/pom.xml
@@ -36,6 +36,10 @@
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-buffer</artifactId>
+ </dependency>
</dependencies>
<build>
diff --git
a/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/protocol/Commands.java
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/protocol/Commands.java
index c5f6c0552..fc833bf73 100644
---
a/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/protocol/Commands.java
+++
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/protocol/Commands.java
@@ -17,51 +17,50 @@
package org.apache.inlong.audit.protocol;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
import org.apache.inlong.audit.protocol.AuditApi.AuditReply;
import org.apache.inlong.audit.protocol.AuditApi.AuditRequest;
import org.apache.inlong.audit.protocol.AuditApi.BaseCommand;
import org.apache.inlong.audit.protocol.AuditApi.BaseCommand.Type;
import org.apache.inlong.audit.protocol.AuditApi.Pong;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.buffer.ChannelBuffers;
public class Commands {
public static int HEAD_LENGTH = 4;
- public static ChannelBuffer getPongChannelBuffer() {
+ public static ByteBuf getPongChannelBuffer() {
BaseCommand cmdPong = BaseCommand.newBuilder()
.setType(Type.PONG)
.setPong(Pong.getDefaultInstance()).build();
return getChannelBuffer(cmdPong.toByteArray());
}
- public static ChannelBuffer getPingChannelBuffer() {
+ public static ByteBuf getPingChannelBuffer() {
BaseCommand cmdPing = BaseCommand.newBuilder()
.setType(Type.PING)
.setPong(Pong.getDefaultInstance()).build();
return getChannelBuffer(cmdPing.toByteArray());
}
- public static ChannelBuffer getAuditRequestBuffer(AuditRequest
auditRequest) {
+ public static ByteBuf getAuditRequestBuffer(AuditRequest auditRequest) {
BaseCommand cmdAuditRequest = BaseCommand.newBuilder()
.setType(Type.AUDITREQUEST)
.setAuditRequest(auditRequest).build();
return getChannelBuffer(cmdAuditRequest.toByteArray());
}
- public static ChannelBuffer getAuditReplylBuffer(AuditReply auditReply) {
+ public static ByteBuf getAuditReplylBuffer(AuditReply auditReply) {
BaseCommand cmdAuditReply = BaseCommand.newBuilder()
.setType(Type.AUDITREPLY)
.setAuditReply(auditReply).build();
return getChannelBuffer(cmdAuditReply.toByteArray());
}
- private static ChannelBuffer getChannelBuffer(byte[] body) {
+ private static ByteBuf getChannelBuffer(byte[] body) {
/* [totalSize] | [body]*/
int totalLength = body.length;
- ChannelBuffer cmdPingBuffer = ChannelBuffers.buffer(
- HEAD_LENGTH + totalLength);
+ ByteBuf cmdPingBuffer = ByteBufAllocator.DEFAULT.buffer(HEAD_LENGTH +
totalLength);
cmdPingBuffer.writeInt(totalLength);
cmdPingBuffer.writeBytes(body);
return cmdPingBuffer;
diff --git a/inlong-audit/audit-proxy/pom.xml b/inlong-audit/audit-proxy/pom.xml
index 13f35eea9..90949f9fb 100644
--- a/inlong-audit/audit-proxy/pom.xml
+++ b/inlong-audit/audit-proxy/pom.xml
@@ -31,15 +31,6 @@
<groupId>org.apache.inlong</groupId>
<artifactId>audit-proxy</artifactId>
- <properties>
- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- <compiler.source>1.8</compiler.source>
- <compiler.target>1.8</compiler.target>
- <junit.version>4.13</junit.version>
- <guava.version>19.0</guava.version>
- <skipTests>false</skipTests>
- </properties>
-
<dependencies>
<dependency>
<groupId>org.apache.inlong</groupId>
@@ -47,26 +38,32 @@
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava.version}</version>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport</artifactId>
</dependency>
<dependency>
- <groupId>org.powermock</groupId>
- <artifactId>powermock-module-junit4</artifactId>
- <version>2.0.2</version>
- <scope>test</scope>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-handler</artifactId>
</dependency>
<dependency>
- <groupId>org.powermock</groupId>
- <artifactId>powermock-api-mockito2</artifactId>
- <version>2.0.2</version>
- <scope>test</scope>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport-native-epoll</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
- <version>${junit.version}</version>
<scope>test</scope>
</dependency>
<dependency>
@@ -81,7 +78,6 @@
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-client</artifactId>
- <version>${project.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
@@ -89,5 +85,21 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <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-api</artifactId>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
index 98a2b9f65..8dc4c644a 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
@@ -19,6 +19,7 @@ package org.apache.inlong.audit.sink;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.RateLimiter;
+import io.netty.handler.codec.TooLongFrameException;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
@@ -42,7 +43,6 @@ import
org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException
import
org.apache.pulsar.client.api.PulsarClientException.NotConnectedException;
import
org.apache.pulsar.client.api.PulsarClientException.ProducerQueueIsFullError;
import
org.apache.pulsar.client.api.PulsarClientException.TopicTerminatedException;
-import org.jboss.netty.handler.codec.frame.TooLongFrameException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/DefaultServiceDecoder.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/DefaultServiceDecoder.java
index 23bb2a209..3f419ca77 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/DefaultServiceDecoder.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/DefaultServiceDecoder.java
@@ -17,9 +17,9 @@
package org.apache.inlong.audit.source;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
import org.apache.inlong.audit.consts.ConfigConstants;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.inlong.audit.protocol.AuditApi.BaseCommand;
@@ -29,7 +29,7 @@ public class DefaultServiceDecoder implements ServiceDecoder {
private static final Logger LOG =
LoggerFactory.getLogger(DefaultServiceDecoder.class);
@Override
- public BaseCommand extractData(ChannelBuffer cb, Channel channel) throws
Exception {
+ public BaseCommand extractData(ByteBuf cb, Channel channel) throws
Exception {
/*[cmd size] | [cmd]*/
if (null == cb) {
LOG.error("cb == null");
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java
index e254c6e01..8b66cd365 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java
@@ -17,23 +17,20 @@
package org.apache.inlong.audit.source;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import io.netty.handler.timeout.ReadTimeoutHandler;
import java.lang.reflect.Constructor;
import java.util.concurrent.TimeUnit;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.source.AbstractSource;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.jboss.netty.channel.Channels;
-import org.jboss.netty.channel.SimpleChannelHandler;
-import org.jboss.netty.channel.group.ChannelGroup;
-import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
-import org.jboss.netty.handler.timeout.ReadTimeoutHandler;
-import org.jboss.netty.util.HashedWheelTimer;
-import org.jboss.netty.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class ServerMessageFactory implements ChannelPipelineFactory {
+public class ServerMessageFactory extends ChannelInitializer<SocketChannel> {
private static final Logger LOG =
LoggerFactory.getLogger(ServerMessageFactory.class);
private static final int DEFAULT_READ_IDLE_TIME = 70 * 60 * 1000;
@@ -45,7 +42,6 @@ public class ServerMessageFactory implements
ChannelPipelineFactory {
private int maxConnections = Integer.MAX_VALUE;
private int maxMsgLength;
private String name;
- private Timer timer = new HashedWheelTimer();
/**
* get server factory
@@ -72,39 +68,29 @@ public class ServerMessageFactory implements
ChannelPipelineFactory {
}
@Override
- public ChannelPipeline getPipeline() throws Exception {
- ChannelPipeline cp = Channels.pipeline();
- return addMessageHandlersTo(cp);
- }
-
- /**
- * get message handlers
- * @param cp
- * @return
- */
- public ChannelPipeline addMessageHandlersTo(ChannelPipeline cp) {
-
- cp.addLast("messageDecoder", new LengthFieldBasedFrameDecoder(
- this.maxMsgLength, 0, 4, 0, 0, true));
- cp.addLast("readTimeoutHandler", new ReadTimeoutHandler(timer,
- DEFAULT_READ_IDLE_TIME, TimeUnit.MILLISECONDS));
+ protected void initChannel(SocketChannel ch) throws Exception {
+ ch.pipeline().addLast("messageDecoder",
+ new LengthFieldBasedFrameDecoder(this.maxMsgLength,
+ 0, 4, 0, 0, true));
+ ch.pipeline().addLast("readTimeoutHandler",
+ new ReadTimeoutHandler(DEFAULT_READ_IDLE_TIME,
TimeUnit.MILLISECONDS));
if (processor != null) {
try {
- Class<? extends SimpleChannelHandler> clazz = (Class<? extends
SimpleChannelHandler>) Class
- .forName(messageHandlerName);
+ Class<? extends ChannelInboundHandlerAdapter> clazz =
+ (Class<? extends ChannelInboundHandlerAdapter>)
Class.forName(messageHandlerName);
- Constructor<?> ctor = clazz.getConstructor(
- AbstractSource.class, ServiceDecoder.class,
ChannelGroup.class, Integer.class);
+ Constructor<?> ctor =
+ clazz.getConstructor(AbstractSource.class,
ServiceDecoder.class,
+ ChannelGroup.class, Integer.class);
- SimpleChannelHandler messageHandler = (SimpleChannelHandler)
ctor
+ ChannelInboundHandlerAdapter messageHandler =
(ChannelInboundHandlerAdapter) ctor
.newInstance(source, serviceDecoder, allChannels,
maxConnections);
- cp.addLast("messageHandler", messageHandler);
+ ch.pipeline().addLast("messageHandler", messageHandler);
} catch (Exception e) {
LOG.info("SimpleChannelHandler.newInstance has error:" +
name, e);
}
}
- return cp;
}
}
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageHandler.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageHandler.java
index c0d372392..93f0423a7 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageHandler.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageHandler.java
@@ -20,9 +20,13 @@ package org.apache.inlong.audit.source;
import static com.google.common.base.Preconditions.checkArgument;
import com.google.gson.Gson;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.group.ChannelGroup;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
-import java.net.SocketAddress;
import java.util.List;
import org.apache.flume.Event;
import org.apache.flume.channel.ChannelProcessor;
@@ -36,14 +40,6 @@ import
org.apache.inlong.audit.protocol.AuditApi.AuditRequest;
import org.apache.inlong.audit.protocol.AuditApi.BaseCommand;
import org.apache.inlong.audit.protocol.AuditData;
import org.apache.inlong.audit.protocol.Commands;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.channel.Channel;
-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.SimpleChannelHandler;
-import org.jboss.netty.channel.group.ChannelGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,7 +47,7 @@ import org.slf4j.LoggerFactory;
* Server message handler
*
*/
-public class ServerMessageHandler extends SimpleChannelHandler {
+public class ServerMessageHandler extends ChannelInboundHandlerAdapter {
private static final Logger logger =
LoggerFactory.getLogger(ServerMessageHandler.class);
@@ -77,32 +73,39 @@ public class ServerMessageHandler extends
SimpleChannelHandler {
}
@Override
- public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
if (allChannels.size() - 1 >= maxConnections) {
logger.warn("refuse to connect , and connections=" +
(allChannels.size() - 1)
+ ", maxConnections="
- + maxConnections + ",channel is " + e.getChannel());
- e.getChannel().disconnect();
- e.getChannel().close();
+ + maxConnections + ",channel is " + ctx.channel());
+ ctx.channel().disconnect();
+ ctx.channel().close();
}
+ allChannels.add(ctx.channel());
+ ctx.fireChannelActive();
}
@Override
- public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
+ public void channelInactive(ChannelHandlerContext ctx) {
+ ctx.fireChannelInactive();
+ allChannels.remove(ctx.channel());
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws
Exception {
logger.debug("message received");
- if (e == null) {
+ if (msg == null) {
logger.warn("get null message event, just skip");
return;
}
- ChannelBuffer cb = ((ChannelBuffer) e.getMessage());
- SocketAddress remoteSocketAddress = e.getRemoteAddress();
+ ByteBuf cb = (ByteBuf) msg;
int len = cb.readableBytes();
if (len == 0) {
logger.warn("receive message skip empty msg.");
cb.clear();
return;
}
- Channel remoteChannel = e.getChannel();
+ Channel remoteChannel = ctx.channel();
BaseCommand cmd = null;
try {
cmd = serviceDecoder.extractData(cb, remoteChannel);
@@ -115,7 +118,7 @@ public class ServerMessageHandler extends
SimpleChannelHandler {
logger.warn("receive message extractData is null");
return;
}
- ChannelBuffer channelBuffer = null;
+ ByteBuf channelBuffer = null;
switch (cmd.getType()) {
case PING:
checkArgument(cmd.hasPing());
@@ -135,7 +138,7 @@ public class ServerMessageHandler extends
SimpleChannelHandler {
break;
}
if (channelBuffer != null) {
- writeResponse(remoteChannel, remoteSocketAddress, channelBuffer);
+ writeResponse(remoteChannel, channelBuffer);
}
}
@@ -194,19 +197,13 @@ public class ServerMessageHandler extends
SimpleChannelHandler {
}
@Override
- public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
throws Exception {
- logger.error("exception caught", e.getCause());
- }
-
- @Override
- public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
- logger.error("channel closed {}", ctx.getChannel());
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
+ logger.error("exception caught", cause);
}
- private void writeResponse(Channel remoteChannel,
- SocketAddress remoteSocketAddress, ChannelBuffer buffer) throws
Exception {
+ private void writeResponse(Channel remoteChannel, ByteBuf buffer) throws
Exception {
if (remoteChannel.isWritable()) {
- remoteChannel.write(buffer, remoteSocketAddress);
+ remoteChannel.writeAndFlush(buffer);
} else {
logger.warn(
"the send buffer2 is full, so disconnect it!please check
remote client"
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServiceDecoder.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServiceDecoder.java
index 786b2cf4e..0ced156cf 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServiceDecoder.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServiceDecoder.java
@@ -17,9 +17,9 @@
package org.apache.inlong.audit.source;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
import org.apache.inlong.audit.protocol.AuditApi.BaseCommand;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.channel.Channel;
/**
* decoder interface definition
@@ -29,5 +29,5 @@ public interface ServiceDecoder {
/**
* extract data from buffer and convert it into map.
*/
- BaseCommand extractData(ChannelBuffer cb, Channel channel) throws
Exception;
+ BaseCommand extractData(ByteBuf cb, Channel channel) throws Exception;
}
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
index 95121bd6b..6a5afdb95 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
@@ -18,9 +18,19 @@
package org.apache.inlong.audit.source;
import com.google.common.base.Preconditions;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import io.netty.util.concurrent.GlobalEventExecutor;
import java.lang.reflect.Constructor;
import java.net.InetSocketAddress;
-import java.util.concurrent.Executors;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
@@ -28,19 +38,10 @@ import org.apache.flume.EventDrivenSource;
import org.apache.flume.FlumeException;
import org.apache.flume.conf.Configurable;
import org.apache.flume.source.AbstractSource;
-import org.apache.inlong.audit.base.NamedThreadFactory;
import org.apache.inlong.audit.channel.FailoverChannelProcessor;
import org.apache.inlong.audit.consts.ConfigConstants;
+import org.apache.inlong.audit.utils.EventLoopUtil;
import org.apache.inlong.audit.utils.FailoverChannelProcessorHolder;
-import org.jboss.netty.bootstrap.ServerBootstrap;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelFactory;
-import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.jboss.netty.channel.group.ChannelGroup;
-import org.jboss.netty.channel.group.DefaultChannelGroup;
-import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
-import org.jboss.netty.util.ThreadNameDeterminer;
-import org.jboss.netty.util.ThreadRenamingRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,8 +56,14 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
protected int maxConnections = Integer.MAX_VALUE;
protected Context context;
- private ServerBootstrap serverBootstrap = null;
+
+ private ServerBootstrap bootstrap = null;
+ protected EventLoopGroup acceptorGroup;
+ protected EventLoopGroup workerGroup;
+ protected DefaultThreadFactory acceptorThreadFactory;
protected ChannelGroup allChannels;
+ protected ChannelFuture channelFuture;
+
protected int port;
protected String host = null;
protected String msgFactoryName;
@@ -94,11 +101,16 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
private static int SEND_BUFFER_MAX_SIZE = 16 * 1024 * 1024;
+ protected boolean enableBusyWait = false;
+
+ protected int acceptorThreads = 1;
+
private Channel nettyChannel = null;
public SimpleTcpSource() {
super();
- allChannels = new DefaultChannelGroup();
+ allChannels = new DefaultChannelGroup("DefaultAuditChannelGroup",
+ GlobalEventExecutor.INSTANCE);
}
@Override
@@ -113,40 +125,42 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
}
super.start();
-
ThreadRenamingRunnable.setThreadNameDeterminer(ThreadNameDeterminer.CURRENT);
- ChannelFactory factory = new NioServerSocketChannelFactory(Executors
- .newCachedThreadPool(
- new
NamedThreadFactory("tcpSource-nettyBoss-threadGroup")),
- 1,
- Executors.newCachedThreadPool(
- new
NamedThreadFactory("tcpSource-nettyWorker-threadGroup")),
- maxThreads);
- logger.info("Set max workers : {} ;", maxThreads);
- ChannelPipelineFactory fac = null;
-
- serverBootstrap = new ServerBootstrap(factory);
- serverBootstrap.setOption("child.tcpNoDelay", tcpNoDelay);
- serverBootstrap.setOption("child.keepAlive", keepAlive);
- serverBootstrap.setOption("child.receiveBufferSize",
receiveBufferSize);
- serverBootstrap.setOption("child.sendBufferSize", sendBufferSize);
- serverBootstrap.setOption("child.writeBufferHighWaterMark",
highWaterMark);
+ acceptorThreadFactory = new
DefaultThreadFactory("tcpSource-nettyBoss-threadGroup");
+
+ this.acceptorGroup = EventLoopUtil.newEventLoopGroup(
+ acceptorThreads, false, acceptorThreadFactory);
+
+ this.workerGroup = EventLoopUtil
+ .newEventLoopGroup(maxThreads, enableBusyWait,
+ new
DefaultThreadFactory("tcpSource-nettyWorker-threadGroup"));
+
+ bootstrap = new ServerBootstrap();
+ bootstrap.childOption(ChannelOption.ALLOCATOR,
ByteBufAllocator.DEFAULT);
+ bootstrap.childOption(ChannelOption.TCP_NODELAY, tcpNoDelay);
+ bootstrap.childOption(ChannelOption.SO_KEEPALIVE, keepAlive);
+ bootstrap.childOption(ChannelOption.SO_RCVBUF, receiveBufferSize);
+ bootstrap.childOption(ChannelOption.SO_SNDBUF, sendBufferSize);
+ bootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK,
highWaterMark);
+
+
bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(workerGroup));
+ EventLoopUtil.enableTriggeredMode(bootstrap);
+ bootstrap.group(acceptorGroup, workerGroup);
+
logger.info("load msgFactory=" + msgFactoryName + " and
serviceDecoderName="
+ serviceDecoderName);
- try {
+ ChannelInitializer fac = null;
+ try {
ServiceDecoder serviceDecoder =
(ServiceDecoder)
Class.forName(serviceDecoderName).newInstance();
-
- Class<? extends ChannelPipelineFactory> clazz =
- (Class<? extends ChannelPipelineFactory>)
Class.forName(msgFactoryName);
-
+ Class<? extends ChannelInitializer> clazz =
+ (Class<? extends ChannelInitializer>)
Class.forName(msgFactoryName);
Constructor ctor =
clazz.getConstructor(AbstractSource.class,
ChannelGroup.class,
ServiceDecoder.class, String.class,
Integer.class, Integer.class, String.class);
-
logger.info("Using channel processor:{}",
this.getClass().getName());
- fac = (ChannelPipelineFactory) ctor
+ fac = (ChannelInitializer) ctor
.newInstance(this, allChannels, serviceDecoder,
messageHandlerName, maxMsgLength, maxConnections,
this.getName());
@@ -157,25 +171,20 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
stop();
throw new FlumeException(e.getMessage());
}
-
- serverBootstrap.setPipelineFactory(fac);
+ bootstrap.childHandler(fac);
try {
if (host == null) {
- nettyChannel = serverBootstrap.bind(new
InetSocketAddress(port));
+ channelFuture = bootstrap.bind(new
InetSocketAddress(port)).sync();
} else {
- nettyChannel = serverBootstrap.bind(new
InetSocketAddress(host, port));
+ channelFuture = bootstrap.bind(new InetSocketAddress(host,
port)).sync();
}
} catch (Exception e) {
logger.error("Simple TCP Source error bind host {} port {},program
will exit!", host,
port);
System.exit(-1);
}
-
- allChannels.add(nettyChannel);
-
logger.info("Simple TCP Source started at host {}, port {}", host,
port);
-
}
@Override
@@ -183,7 +192,6 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
logger.info("[STOP SOURCE]{} stopping...", super.getName());
if (allChannels != null && !allChannels.isEmpty()) {
try {
- allChannels.unbind().awaitUninterruptibly();
allChannels.close().awaitUninterruptibly();
} catch (Exception e) {
logger.warn("Simple TCP Source netty server stop ex", e);
@@ -193,17 +201,6 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
}
}
- if (serverBootstrap != null) {
- try {
-
- serverBootstrap.releaseExternalResources();
- } catch (Exception e) {
- logger.warn("Simple TCP Source serverBootstrap stop ex ", e);
- } finally {
- serverBootstrap = null;
- }
- }
-
super.stop();
logger.info("[STOP SOURCE]{} stopped", super.getName());
}
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/utils/EventLoopUtil.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/utils/EventLoopUtil.java
new file mode 100644
index 000000000..8f43532bd
--- /dev/null
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/utils/EventLoopUtil.java
@@ -0,0 +1,114 @@
+/*
+ * 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.audit.utils;
+
+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<Void> by Future
+ *
+ * @param future Future
+ * @return CompletableFuture<Void>
+ */
+ 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;
+ }
+}
diff --git a/inlong-audit/audit-sdk/pom.xml b/inlong-audit/audit-sdk/pom.xml
index 2c8d6558c..edd4b32e9 100644
--- a/inlong-audit/audit-sdk/pom.xml
+++ b/inlong-audit/audit-sdk/pom.xml
@@ -38,14 +38,28 @@
<junit.version>4.13</junit.version>
<commons.version>3.0</commons.version>
<skipTests>false</skipTests>
- <slf4j.version>1.7.25</slf4j.version>
</properties>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
- <artifactId>netty</artifactId>
- <version>${netty.version}</version>
+ <artifactId>netty-transport</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-handler</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport-native-epoll</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
@@ -64,11 +78,5 @@
<version>${project.version}</version>
<scope>compile</scope>
</dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-simple</artifactId>
- <version>${slf4j.version}</version>
- <scope>compile</scope>
- </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/ClientPipelineFactory.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/ClientPipelineFactory.java
new file mode 100644
index 000000000..5f22c1793
--- /dev/null
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/ClientPipelineFactory.java
@@ -0,0 +1,39 @@
+/*
+ * 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.audit.send;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import org.apache.inlong.audit.util.Decoder;
+
+public class ClientPipelineFactory extends ChannelInitializer<SocketChannel> {
+ private final SimpleChannelInboundHandler sendHandler;
+
+ public ClientPipelineFactory(SimpleChannelInboundHandler sendHandler) {
+ this.sendHandler = sendHandler;
+ }
+
+ @Override
+ public void initChannel(SocketChannel ch) throws Exception {
+ ch.pipeline().addLast("contentDecoder", new Decoder());
+ ch.pipeline().addLast("handler", sendHandler);
+ }
+}
+
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderChannel.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderChannel.java
index 1ecbd4fc1..224e5d2e5 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderChannel.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderChannel.java
@@ -18,8 +18,7 @@
package org.apache.inlong.audit.send;
import org.apache.inlong.audit.util.IpPort;
-import org.jboss.netty.channel.Channel;
-
+import io.netty.channel.Channel;
import java.util.concurrent.Semaphore;
public class SenderChannel {
@@ -38,7 +37,6 @@ public class SenderChannel {
this.channel = channel;
this.ipPort = ipPort;
this.packToken = new Semaphore(maxSynchRequest);
- this.channel.getConfig().setConnectTimeoutMillis(5000);
}
/**
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderGroup.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderGroup.java
index c578f069f..793b99308 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderGroup.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderGroup.java
@@ -17,18 +17,18 @@
package org.apache.inlong.audit.send;
-import org.apache.inlong.audit.util.Encoder;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.ThreadFactory;
+import org.apache.inlong.audit.util.EventLoopUtil;
import org.apache.inlong.audit.util.IpPort;
import org.apache.inlong.audit.util.SenderResult;
-import org.jboss.netty.bootstrap.ClientBootstrap;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelFuture;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelUpstreamHandler;
-import org.jboss.netty.channel.Channels;
-import org.jboss.netty.channel.SimpleChannelHandler;
-import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,7 +38,6 @@ import java.util.List;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class SenderGroup {
@@ -48,8 +47,10 @@ public class SenderGroup {
public static final int DEFAULT_WAIT_TIMES = 10000;
public static final int WAIT_INTERVAL = 1;
public static final int DEFAULT_SYNCH_REQUESTS = 1;
+ public static final int DEFAULT_RECEIVE_BUFFER_SIZE = 16777216;
+ public static final int DEFAULT_SEND_BUFFER_SIZE = 16777216;
- private ClientBootstrap client = new ClientBootstrap();
+ private Bootstrap client;
private List<LinkedBlockingQueue<SenderChannel>> channelGroups = new
ArrayList<>();
private int mIndex = 0;
private List<SenderChannel> deleteChannels = new ArrayList<>();
@@ -65,31 +66,29 @@ public class SenderGroup {
* constructor
*
* @param senderThreadNum
- * @param decoder
* @param clientHandler
*/
- public SenderGroup(int senderThreadNum, ChannelUpstreamHandler decoder,
- SimpleChannelHandler clientHandler) {
+ public SenderGroup(int senderThreadNum, SimpleChannelInboundHandler
clientHandler) {
this.senderThreadNum = senderThreadNum;
- client.setFactory(new NioClientSocketChannelFactory(
- Executors.newCachedThreadPool(),
- Executors.newCachedThreadPool(),
- this.senderThreadNum));
+ ThreadFactory selfDefineFactory = new
DefaultThreadFactory("audit-client-io",
+ Thread.currentThread().isDaemon());
- client.setPipelineFactory(() -> {
- ChannelPipeline pipeline = Channels.pipeline();
- pipeline.addLast("decoder", decoder);
- pipeline.addLast("encoder", new Encoder());
- pipeline.addLast("handler", clientHandler);
- return pipeline;
- });
- client.setOption("tcpNoDelay", true);
- client.setOption("child.tcpNoDelay", true);
- client.setOption("keepAlive", true);
- client.setOption("child.keepAlive", true);
- client.setOption("reuseAddr", true);
+ EventLoopGroup eventLoopGroup =
EventLoopUtil.newEventLoopGroup(this.senderThreadNum,
+ false, selfDefineFactory);
+ client = new Bootstrap();
+ client.group(eventLoopGroup);
+
client.channel(EventLoopUtil.getClientSocketChannelClass(eventLoopGroup));
+ client.option(ChannelOption.SO_KEEPALIVE, true);
+ client.option(ChannelOption.TCP_NODELAY, true);
+ client.option(ChannelOption.SO_REUSEADDR, true);
+ client.option(ChannelOption.SO_RCVBUF, DEFAULT_RECEIVE_BUFFER_SIZE);
+ client.option(ChannelOption.SO_SNDBUF, DEFAULT_SEND_BUFFER_SIZE);
+ client.handler(new ClientPipelineFactory(clientHandler));
+ /*
+ * init add two list for update config
+ */
channelGroups.add(new LinkedBlockingQueue<>());
channelGroups.add(new LinkedBlockingQueue<>());
}
@@ -100,7 +99,7 @@ public class SenderGroup {
* @param dataBuf
* @return
*/
- public SenderResult send(ChannelBuffer dataBuf) {
+ public SenderResult send(ByteBuf dataBuf) {
LinkedBlockingQueue<SenderChannel> channels =
channelGroups.get(mIndex);
SenderChannel channel = null;
try {
@@ -113,16 +112,10 @@ public class SenderGroup {
channels = channelGroups.get(mIndex);
for (int i = 0; i < channels.size(); i++) {
channel = channels.poll();
- boolean ret = channel.tryAcquire();
if (channel.tryAcquire()) {
isOk = true;
break;
}
-
- if (ret) {
- isOk = true;
- break;
- }
channels.offer(channel);
channel = null;
}
@@ -140,17 +133,17 @@ public class SenderGroup {
return new SenderResult("can not get a channel", 0, false);
}
ChannelFuture t = null;
- if (channel.getChannel().isConnected()) {
- t = channel.getChannel().write(dataBuf).sync().await();
+ if (channel.getChannel().isWritable()) {
+ t = channel.getChannel().writeAndFlush(dataBuf).sync().await();
if (!t.isSuccess()) {
- if (!channel.getChannel().isConnected()) {
+ if (!channel.getChannel().isActive()) {
reconnect(channel);
}
- t = channel.getChannel().write(dataBuf).sync().await();
+ t =
channel.getChannel().writeAndFlush(dataBuf).sync().await();
}
} else {
reconnect(channel);
- t = channel.getChannel().write(dataBuf).sync().await();
+ t = channel.getChannel().writeAndFlush(dataBuf).sync().await();
}
return new SenderResult(channel.getIpPort().ip,
channel.getIpPort().port, t.isSuccess());
} catch (Throwable ex) {
@@ -215,7 +208,7 @@ public class SenderGroup {
continue;
}
ChannelFuture future =
client.connect(ipPortObj.addr).await();
- channel = new SenderChannel(future.getChannel(),
ipPortObj, maxSynchRequest);
+ channel = new SenderChannel(future.channel(), ipPortObj,
maxSynchRequest);
newChannels.add(channel);
totalChannels.put(ipPort, channel);
} catch (Exception e) {
@@ -251,7 +244,7 @@ public class SenderGroup {
Channel oldChannel = channel.getChannel();
ChannelFuture future =
client.connect(channel.getIpPort().addr).await();
- Channel newChannel = future.getChannel();
+ Channel newChannel = future.channel();
channel.setChannel(newChannel);
oldChannel.disconnect();
oldChannel.close();
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderHandler.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderHandler.java
index cf43c184e..2a336f349 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderHandler.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderHandler.java
@@ -17,15 +17,13 @@
package org.apache.inlong.audit.send;
-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.SimpleChannelHandler;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class SenderHandler extends SimpleChannelHandler {
+public class SenderHandler extends SimpleChannelInboundHandler<ByteBuf> {
private static final Logger logger =
LoggerFactory.getLogger(SenderHandler.class);
private SenderManager manager;
@@ -42,7 +40,7 @@ public class SenderHandler extends SimpleChannelHandler {
* Message Received
*/
@Override
- public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
+ public void channelRead0(io.netty.channel.ChannelHandlerContext ctx,
ByteBuf e) {
try {
manager.onMessageReceived(ctx, e);
} catch (Throwable ex) {
@@ -54,7 +52,7 @@ public class SenderHandler extends SimpleChannelHandler {
* Caught exception
*/
@Override
- public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) {
try {
manager.onExceptionCaught(ctx, e);
} catch (Throwable ex) {
@@ -66,9 +64,9 @@ public class SenderHandler extends SimpleChannelHandler {
* Disconnected channel
*/
@Override
- public void channelDisconnected(ChannelHandlerContext ctx,
ChannelStateEvent e) {
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
try {
- super.channelDisconnected(ctx, e);
+ super.channelInactive(ctx);
} catch (Throwable ex) {
logger.error(ex.getMessage());
}
@@ -78,9 +76,9 @@ public class SenderHandler extends SimpleChannelHandler {
* Closed channel
*/
@Override
- public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) {
+ public void channelUnregistered(ChannelHandlerContext ctx) throws
Exception {
try {
- super.channelClosed(ctx, e);
+ super.channelUnregistered(ctx);
} catch (Throwable ex) {
logger.error(ex.getMessage());
}
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java
index 49f2596d5..4355a6a09 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java
@@ -17,16 +17,13 @@
package org.apache.inlong.audit.send;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelHandlerContext;
import org.apache.inlong.audit.protocol.AuditApi;
import org.apache.inlong.audit.util.AuditConfig;
import org.apache.inlong.audit.util.AuditData;
-import org.apache.inlong.audit.util.Decoder;
import org.apache.inlong.audit.util.SenderResult;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.channel.ExceptionEvent;
-import org.jboss.netty.channel.MessageEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -84,7 +81,7 @@ public class SenderManager {
this.auditConfig = config;
this.maxConnectChannels = maxConnectChannels;
SenderHandler clientHandler = new SenderHandler(this);
- this.sender = new SenderGroup(DEFAULT_SEND_THREADNUM, new
Decoder(), clientHandler);
+ this.sender = new SenderGroup(DEFAULT_SEND_THREADNUM,
clientHandler);
} catch (Exception ex) {
logger.error(ex.getMessage());
}
@@ -153,7 +150,12 @@ public class SenderManager {
* @param data
*/
private void sendData(byte[] data) {
- ChannelBuffer dataBuf = ChannelBuffers.wrappedBuffer(data);
+ if (data == null || data.length <= 0) {
+ logger.warn("send data is empty!");
+ return;
+ }
+ ByteBuf dataBuf = ByteBufAllocator.DEFAULT.buffer(data.length);
+ dataBuf.writeBytes(data);
SenderResult result = this.sender.send(dataBuf);
if (!result.result) {
this.sender.setHasSendError(true);
@@ -263,18 +265,14 @@ public class SenderManager {
/**
* processing return package
*
- * @param ctx
- * @param e
+ * @param ctx ctx
+ * @param msg msg
*/
- public void onMessageReceived(ChannelHandlerContext ctx, MessageEvent e) {
+ public void onMessageReceived(ChannelHandlerContext ctx, ByteBuf msg) {
try {
//Analyze abnormal events
- if (!(e.getMessage() instanceof ChannelBuffer)) {
- logger.error("onMessageReceived e.getMessage:" +
e.getMessage());
- return;
- }
- ChannelBuffer readBuffer = (ChannelBuffer) e.getMessage();
- byte[] readBytes = readBuffer.toByteBuffer().array();
+ ByteBuf readBuffer = msg;
+ byte[] readBytes = readBuffer.array();
AuditApi.BaseCommand baseCommand =
AuditApi.BaseCommand.parseFrom(readBytes);
// Parse request id
Long requestId = baseCommand.getAuditReply().getRequestId();
@@ -304,7 +302,7 @@ public class SenderManager {
* @param ctx
* @param e
*/
- public void onExceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
{
+ public void onExceptionCaught(ChannelHandlerContext ctx, Throwable e) {
logger.error(e.getCause().getMessage());
try {
this.sender.setHasSendError(true);
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Decoder.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Decoder.java
index 01ea67b83..03ff94312 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Decoder.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Decoder.java
@@ -17,14 +17,13 @@
package org.apache.inlong.audit.util;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.buffer.DynamicChannelBuffer;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.handler.codec.frame.FrameDecoder;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import java.util.List;
-public class Decoder extends FrameDecoder {
+public class Decoder extends MessageToMessageDecoder<ByteBuf> {
// Maximum return packet size
private static final int MAX_RESPONSE_LENGTH = 8 * 1024 * 1024;
@@ -32,7 +31,8 @@ public class Decoder extends FrameDecoder {
* decoding
*/
@Override
- protected Object decode(ChannelHandlerContext ctx, Channel channel,
ChannelBuffer buffer) {
+ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer,
+ List<Object> out) throws Exception {
// Every time you need to read the complete package (that is, read to
the end of the package),
// otherwise only the first one will be parsed correctly,
// which will adversely affect the parsing of the subsequent package
@@ -41,16 +41,15 @@ public class Decoder extends FrameDecoder {
int totalLen = buffer.readInt();
// Respond to abnormal channel, interrupt in time to avoid stuck
if (totalLen > MAX_RESPONSE_LENGTH) {
- channel.close();
- return null;
+ ctx.channel().close();
+ return;
}
// If the package is not complete, continue to wait for the return
package
if (buffer.readableBytes() < totalLen) {
buffer.resetReaderIndex();
- return null;
+ return;
}
- ChannelBuffer returnBuffer = new
DynamicChannelBuffer(ChannelBuffers.BIG_ENDIAN, totalLen);
+ ByteBuf returnBuffer = ByteBufAllocator.DEFAULT.buffer(totalLen);
buffer.readBytes(returnBuffer, 0, totalLen);
- return returnBuffer;
}
}
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Encoder.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Encoder.java
deleted file mode 100644
index abb327085..000000000
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Encoder.java
+++ /dev/null
@@ -1,39 +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.audit.util;
-
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
-
-public class Encoder extends OneToOneEncoder {
-
- /**
- * Encoder
- */
- @Override
- protected Object encode(ChannelHandlerContext ctx, Channel channel, Object
msg) {
- if (!(msg instanceof ChannelBuffer)) {
- return msg;
- } else {
- ChannelBuffer src = (ChannelBuffer) msg;
- return src;
- }
- }
-}
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/EventLoopUtil.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/EventLoopUtil.java
new file mode 100644
index 000000000..d93b6d1ea
--- /dev/null
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/EventLoopUtil.java
@@ -0,0 +1,114 @@
+/*
+ * 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.audit.util;
+
+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<Void> by Future
+ *
+ * @param future Future
+ * @return CompletableFuture<Void>
+ */
+ 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;
+ }
+}
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/IpPort.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/IpPort.java
index fa4ba5eaf..494864b60 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/IpPort.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/IpPort.java
@@ -17,8 +17,8 @@
package org.apache.inlong.audit.util;
+import io.netty.channel.Channel;
import org.apache.commons.lang3.math.NumberUtils;
-import org.jboss.netty.channel.Channel;
import java.net.InetSocketAddress;
@@ -92,10 +92,10 @@ public class IpPort {
*/
public static InetSocketAddress parseInetSocketAddress(Channel channel) {
InetSocketAddress destAddr = null;
- if (channel.getRemoteAddress() instanceof InetSocketAddress) {
- destAddr = (InetSocketAddress) channel.getRemoteAddress();
+ if (channel.remoteAddress() instanceof InetSocketAddress) {
+ destAddr = (InetSocketAddress) channel.remoteAddress();
} else {
- String sendIp = channel.getRemoteAddress().toString();
+ String sendIp = channel.remoteAddress().toString();
destAddr = new InetSocketAddress(sendIp, 0);
}
return destAddr;
diff --git
a/inlong-audit/audit-sdk/src/test/java/org/apache/inlong/audit/send/SenderGroupTest.java
b/inlong-audit/audit-sdk/src/test/java/org/apache/inlong/audit/send/SenderGroupTest.java
index 4cc8192ad..d375a457b 100644
---
a/inlong-audit/audit-sdk/src/test/java/org/apache/inlong/audit/send/SenderGroupTest.java
+++
b/inlong-audit/audit-sdk/src/test/java/org/apache/inlong/audit/send/SenderGroupTest.java
@@ -18,7 +18,6 @@
package org.apache.inlong.audit.send;
import org.apache.inlong.audit.util.AuditConfig;
-import org.apache.inlong.audit.util.Decoder;
import org.junit.Test;
import static org.junit.Assert.assertFalse;
@@ -28,7 +27,7 @@ public class SenderGroupTest {
AuditConfig testConfig = new AuditConfig();
SenderManager testManager = new SenderManager(testConfig);
SenderHandler clientHandler = new
org.apache.inlong.audit.send.SenderHandler(testManager);
- SenderGroup sender = new org.apache.inlong.audit.send.SenderGroup(10, new
Decoder(), clientHandler);
+ SenderGroup sender = new org.apache.inlong.audit.send.SenderGroup(10,
clientHandler);
@Test
public void isHasSendError() {
diff --git a/inlong-audit/audit-store/pom.xml b/inlong-audit/audit-store/pom.xml
index 57ea3c276..21567c434 100644
--- a/inlong-audit/audit-store/pom.xml
+++ b/inlong-audit/audit-store/pom.xml
@@ -142,14 +142,6 @@
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
</dependency>
- <dependency>
- <groupId>ch.qos.logback</groupId>
- <artifactId>logback-classic</artifactId>
- </dependency>
- <dependency>
- <groupId>ch.qos.logback</groupId>
- <artifactId>logback-core</artifactId>
- </dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
@@ -166,13 +158,22 @@
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>tubemq-client</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- </exclusions>
+ </dependency>
+ <dependency>
+ <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-api</artifactId>
</dependency>
</dependencies>
<build>
diff --git a/inlong-audit/bin/audit-proxy b/inlong-audit/bin/audit-proxy
index 90d20862e..dc68b5377 100644
--- a/inlong-audit/bin/audit-proxy
+++ b/inlong-audit/bin/audit-proxy
@@ -92,7 +92,7 @@ run_audit() {
if [ ${CLEAN_FLAG} -ne 0 ]; then
set -x
fi
- $EXEC $JAVA_HOME/bin/java $JAVA_OPTS -cp "$AUDIT_CLASSPATH"
-Djava.library.path=$AUDIT_JAVA_LIBRARY_PATH "$AUDIT_APPLICATION_CLASS" $*
+ $EXEC $JAVA_HOME/bin/java $JAVA_OPTS -cp "$AUDIT_CLASSPATH"
-Daudit.log.path=$LOG_PATH -Djava.library.path=$AUDIT_JAVA_LIBRARY_PATH
"$AUDIT_APPLICATION_CLASS" $*
}
################################
@@ -212,6 +212,8 @@ else
AUDIT_CLASSPATH="${AUDIT_HOME}/lib/*"
fi
+LOG_PATH="${AUDIT_HOME}/logs"
+
# find java
if [ -z "${JAVA_HOME}" ] ; then
warn "JAVA_HOME is not set!"
diff --git a/inlong-audit/bin/store-start.sh b/inlong-audit/bin/store-start.sh
index f24a4e0ed..74aeb1759 100644
--- a/inlong-audit/bin/store-start.sh
+++ b/inlong-audit/bin/store-start.sh
@@ -24,6 +24,8 @@ base_dir=$(
pwd
)
+LOG_PATH="${base_dir}/logs"
+
PID=$(ps -ef | grep "audit-store" | grep -v grep | awk '{ print $2}')
LOG_DIR="${basedir}/logs"
@@ -57,7 +59,8 @@ JAVA_OPTS="${JAVA_OPTS} ${HEAP_OPTS}"
SERVERJAR=`ls -lt ${base_dir}/lib |grep audit-store | head -2 | tail -1 | awk
'{print $NF}'`
-nohup $JAVA $JAVA_OPTS -Dloader.path="$base_dir/conf,$base_dir/lib/" -jar
"$base_dir/lib/$SERVERJAR" 1>${LOG_DIR}/store.log 2>${LOG_DIR}/store-error.log
&
+nohup $JAVA $JAVA_OPTS -Daudit.log.path=$LOG_PATH
-Dloader.path="$base_dir/conf,$base_dir/lib/" -jar "$base_dir/lib/$SERVERJAR"
1>${LOG_DIR}/store.log 2>${LOG_DIR}/store-error.log &
+
PIDFILE="$base_dir/bin/PID"
PID=$(ps -ef | grep "$base_dir" | grep -v grep | awk '{ print $2}')
diff --git a/inlong-audit/conf/log4j2.xml b/inlong-audit/conf/log4j2.xml
new file mode 100644
index 000000000..2cdf3b2dd
--- /dev/null
+++ b/inlong-audit/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:audit.log.path}</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}/info.log</property>
+ <property
name="info_filePattern">${basePath}/info-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="info_max">10</property>
+ <property name="debug_fileName">${basePath}/debug.log</property>
+ <property
name="debug_filePattern">${basePath}/debug-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="debug_max">10</property>
+ <property name="warn_fileName">${basePath}/warn.log</property>
+ <property
name="warn_filePattern">${basePath}/warn-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="warn_max">10</property>
+ <property name="error_fileName">${basePath}/error.log</property>
+ <property
name="error_filePattern">${basePath}/error-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="error_max">10</property>
+ <property name="console_print_level">DEBUG</property>
+ <property name="index_fileName">${basePath}/index.log</property>
+ <property
name="index_filePattern">${basePath}/index-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="monitors_fileName">${basePath}/monitors.log</property>
+ <property
name="monitors_filePattern">${basePath}/monitors-%d{yyyy-MM-dd}-%i.log.gz</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="IndexFile" fileName="${index_fileName}"
filePattern="${index_filePattern}">
+ <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %m%n"/>
+ <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="MonitorFile" fileName="${monitors_fileName}"
filePattern="${monitors_filePattern}">
+ <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %m%n"/>
+ <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>
+ </appenders>
+
+ <loggers>
+ <logger name="org.apache.pulsar" level="info" additivity="false">
+ <appender-ref ref="InfoFile"/>
+ </logger>
+ <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-audit/conf/logback.xml b/inlong-audit/conf/logback.xml
deleted file mode 100644
index 1d74464f8..000000000
--- a/inlong-audit/conf/logback.xml
+++ /dev/null
@@ -1,98 +0,0 @@
-<?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 scan="true" scanPeriod="60s">
- <property name="LOG_BACK_DIR" value="logs"/>
-
- <appender name="INFO_ROLLING"
class="ch.qos.logback.core.rolling.RollingFileAppender">
- <file>${LOG_BACK_DIR}/audit.log</file>
- <rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
- <!-- rollover daily -->
-
<fileNamePattern>${LOG_BACK_DIR}/info-%d{yyyy-MM-dd}.%i.gz</fileNamePattern>
- <!-- each file should be at most 100MB, keep 60 days
worth of history, but at most 20GB -->
- <maxFileSize>1024MB</maxFileSize>
- <maxHistory>2</maxHistory>
- <totalSizeCap>50GB</totalSizeCap>
- </rollingPolicy>
- <encoder>
- <pattern>%d{yyy-MM-dd HH:mm:ss.SSS,GMT+8} %p %t -
%m%n</pattern>
- <charset
class="java.nio.charset.Charset">UTF-8</charset>
- </encoder>
- <filter class="ch.qos.logback.classic.filter.LevelFilter">
- <level>INFO</level>
- <onMatch>ACCEPT</onMatch>
- <onMismatch>DENY</onMismatch>
- </filter>
- </appender>
-
- <appender name="ERROR_ROLLING"
class="ch.qos.logback.core.rolling.RollingFileAppender">
- <file>${LOG_BACK_DIR}/error.log</file>
- <rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
- <!-- rollover daily -->
-
<fileNamePattern>${LOG_BACK_DIR}/error-%d{yyyy-MM-dd}.%i.gz</fileNamePattern>
- <!-- each file should be at most 100MB, keep 60 days
worth of history, but at most 20GB -->
- <maxFileSize>1024MB</maxFileSize>
- <maxHistory>2</maxHistory>
- <totalSizeCap>50GB</totalSizeCap>
- </rollingPolicy>
- <encoder>
- <pattern>%d{yyy-MM-dd HH:mm:ss.SSS,GMT+8} %p %t -
%m%n</pattern>
- <charset
class="java.nio.charset.Charset">UTF-8</charset>
- </encoder>
- <filter class="ch.qos.logback.classic.filter.LevelFilter">
- <level>ERROR</level>
- <onMatch>ACCEPT</onMatch>
- <onMismatch>DENY</onMismatch>
- </filter>
- </appender>
-
- <appender name="DEBUG_ROLLING"
class="ch.qos.logback.core.rolling.RollingFileAppender">
- <file>${LOG_BACK_DIR}/debug.log</file>
- <rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
- <!-- rollover daily -->
-
<fileNamePattern>${LOG_BACK_DIR}/debug-%d{yyyy-MM-dd}.%i.gz</fileNamePattern>
- <!-- each file should be at most 100MB, keep 60 days
worth of history, but at most 20GB -->
- <maxFileSize>1024MB</maxFileSize>
- <maxHistory>2</maxHistory>
- <totalSizeCap>50GB</totalSizeCap>
- </rollingPolicy>
- <encoder>
- <pattern>%d{yyy-MM-dd HH:mm:ss.SSS,GMT+8} %p %t -
%m%n</pattern>
- <charset
class="java.nio.charset.Charset">UTF-8</charset>
- </encoder>
- <filter class="ch.qos.logback.classic.filter.LevelFilter">
- <level>DEBUG</level>
- <onMatch>ACCEPT</onMatch>
- <onMismatch>DENY</onMismatch>
- </filter>
- </appender>
-
- <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
- <layout class="ch.qos.logback.classic.PatternLayout">
- <param name="Pattern"
- value="%d{yyyy-MM-dd
HH:mm:ss.SSS}|%X{threadId}|%X{traceId}-%X{rpcId}|%level|%C|%M|%L|uid=%X{uid},deviceId=%X{deviceId},sessionId=%X{sessionId},requestId=%X{requestId},tradeId=%X{tradeId},clientIp=%X{clientIp}|%.-512msg|%X{callChain}%n"/>
- </layout>
- </appender>
-
- <root level="DEBUG">
- <appender-ref ref="INFO_ROLLING"/>
- <appender-ref ref="DEBUG_ROLLING"/>
- <appender-ref ref="ERROR_ROLLING"/>
- </root>
-</configuration>
diff --git a/inlong-audit/pom.xml b/inlong-audit/pom.xml
index 3a6754f0f..0d8453daf 100644
--- a/inlong-audit/pom.xml
+++ b/inlong-audit/pom.xml
@@ -46,6 +46,7 @@
<maven.compiler.target>8</maven.compiler.target>
<flume.version>1.9.0</flume.version>
<plugin.assembly.version>3.2.0</plugin.assembly.version>
+ <netty.version>4.1.72.Final</netty.version>
<codec.version>1.15</codec.version>
<servlet.version>2.5-20110124</servlet.version>
<inlong-common.version>1.3.4</inlong-common.version>
@@ -60,9 +61,12 @@
<elasticsearch.version>6.8.23</elasticsearch.version>
<gson.version>2.8.6</gson.version>
<jackson.version>2.12.6.1</jackson.version>
- <junit.version>4.12</junit.version>
+ <guava.version>19.0</guava.version>
+ <junit.version>4.13</junit.version>
<autoconfigure.version>2.4.3</autoconfigure.version>
<spring.plugin.version>2.6.6</spring.plugin.version>
+ <log4j2.version>2.17.1</log4j2.version>
+ <slf4j.version>1.7.36</slf4j.version>
</properties>
<dependencyManagement>
@@ -91,6 +95,12 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<version>${spring.boot.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-logging</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
@@ -119,6 +129,11 @@
<artifactId>pulsar-client</artifactId>
<version>${pulsar.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>tubemq-client</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
@@ -139,6 +154,11 @@
<artifactId>gson</artifactId>
<version>${gson.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${guava.version}</version>
+ </dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
@@ -159,6 +179,56 @@
<artifactId>testng</artifactId>
<version>${testng.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport</artifactId>
+ <version>${netty.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-handler</artifactId>
+ <version>${netty.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-common</artifactId>
+ <version>${netty.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec</artifactId>
+ <version>${netty.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport-native-epoll</artifactId>
+ <version>${netty.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-buffer</artifactId>
+ <version>${netty.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <version>${log4j2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <version>${log4j2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <version>${log4j2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>${slf4j.version}</version>
+ </dependency>
</dependencies>
</dependencyManagement>
<dependencies>
@@ -182,11 +252,6 @@
<artifactId>flume-ng-configuration</artifactId>
<version>${flume.version}</version>
</dependency>
- <dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty</artifactId>
- <version>${netty.version}</version>
- </dependency>
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
index ca5070ea1..8c57c532c 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
@@ -87,20 +87,8 @@ public class ServerMessageHandler extends
ChannelInboundHandlerAdapter {
.trimResults().withKeyValueSeparator(AttributeConstants.KEY_VALUE_SEPARATOR);
private static final ThreadLocal<SimpleDateFormat> dateFormator =
- new ThreadLocal<SimpleDateFormat>() {
- @Override
- protected SimpleDateFormat initialValue() {
- return new SimpleDateFormat("yyyyMMddHHmm");
- }
- };
+ ThreadLocal.withInitial(() -> new
SimpleDateFormat("yyyyMMddHHmm"));
- private static final ThreadLocal<SimpleDateFormat> dateFormator4Transfer =
- new ThreadLocal<SimpleDateFormat>() {
- @Override
- protected SimpleDateFormat initialValue() {
- return new SimpleDateFormat("yyyyMMddHHmmss");
- }
- };
private AbstractSource source;
private final ChannelGroup allChannels;
@@ -269,7 +257,7 @@ public class ServerMessageHandler extends
ChannelInboundHandlerAdapter {
}
@Override
- public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ public void channelInactive(ChannelHandlerContext ctx) {
logger.error("channel inactive {}", ctx.channel());
ctx.fireChannelInactive();
allChannels.remove(ctx.channel());
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
index 34af408f3..b31e88e29 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
@@ -133,7 +133,6 @@ public class DefaultMessageSender implements MessageSender {
if (configure.isEnableSaveManagerVIps()
&& configure.isLocalVisit()
-
&& ManagerFetcherThreadStarted.compareAndSet(false, true)) {
managerFetcherThread = new ManagerFetcherThread(configure);
managerFetcherThread.start();
