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/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 32a654b17 [INLONG-7950][DataProxy] Optimize the implementation logic
of the Source (#8041)
32a654b17 is described below
commit 32a654b17dfcba090ab022e8359e49e5e818fc23
Author: Goson Zhang <[email protected]>
AuthorDate: Thu May 18 18:55:18 2023 +0800
[INLONG-7950][DataProxy] Optimize the implementation logic of the Source
(#8041)
---
.../inlong/common/enums/DataProxyErrCode.java | 20 +
.../apache/inlong/dataproxy/base/SinkRspEvent.java | 12 +-
.../inlong/dataproxy/consts/StatConstants.java | 29 +-
.../dataproxy/sink/mq/OrderBatchPackProfileV0.java | 6 +-
.../dataproxy/source/ServerMessageHandler.java | 2 +-
.../inlong/dataproxy/source2/BaseSource.java | 544 ++++++++++++++++
.../dataproxy/source2/InLongMessageFactory.java | 77 +++
.../dataproxy/source2/InLongMessageHandler.java | 681 +++++++++++++++++++++
.../inlong/dataproxy/source2/SimpleTcpSource.java | 153 +++++
.../inlong/dataproxy/source2/SimpleUdpSource.java | 82 +++
.../inlong/dataproxy/source2/SourceConstants.java | 202 ++++++
.../dataproxy/source2/v0msg/AbsV0MsgCodec.java | 223 +++++++
.../dataproxy/source2/v0msg/CodecBinMsg.java | 363 +++++++++++
.../dataproxy/source2/v0msg/CodecTextMsg.java | 257 ++++++++
.../dataproxy/source2/v0msg/MsgFieldConsts.java | 74 +++
.../inlong/dataproxy/utils/AddressUtils.java | 24 +-
.../inlong/dataproxy/utils/MessageUtils.java | 6 +-
17 files changed, 2732 insertions(+), 23 deletions(-)
diff --git
a/inlong-common/src/main/java/org/apache/inlong/common/enums/DataProxyErrCode.java
b/inlong-common/src/main/java/org/apache/inlong/common/enums/DataProxyErrCode.java
index 61953ba99..263af3678 100644
---
a/inlong-common/src/main/java/org/apache/inlong/common/enums/DataProxyErrCode.java
+++
b/inlong-common/src/main/java/org/apache/inlong/common/enums/DataProxyErrCode.java
@@ -17,6 +17,8 @@
package org.apache.inlong.common.enums;
+import org.apache.commons.lang3.math.NumberUtils;
+
/**
* Enum of data proxy error code.
*/
@@ -25,13 +27,23 @@ public enum DataProxyErrCode {
SUCCESS(0, "Ok"),
SINK_SERVICE_UNREADY(1, "Service not ready"),
+ SERVICE_CLOSED(2, "Service closed"),
+ CONF_SERVICE_UNREADY(3, "Configure Service not ready"),
ILLEGAL_VISIT_IP(10, "Illegal visit ip"),
+ FIELD_VALUE_NOT_EQUAL(95, "Field value not equal"),
+ UNCOMPRESS_DATA_ERROR(96, "Uncompress data error"),
+
MISS_REQUIRED_GROUPID_ARGUMENT(100, "Parameter groupId is required"),
MISS_REQUIRED_STREAMID_ARGUMENT(101, "Parameter streamId is required"),
MISS_REQUIRED_DT_ARGUMENT(102, "Parameter dt is required"),
MISS_REQUIRED_BODY_ARGUMENT(103, "Parameter body is required"),
BODY_EXCEED_MAX_LEN(104, "Body length exceed the maximum length"),
+ BODY_LENGTH_ZERO(105, "Body length is 0"),
+ BODY_LENGTH_LESS_ZERO(106, "Body length less than 0"),
+ ATTR_LENGTH_LESS_ZERO(107, "Attribute length less than 0"),
+ ERROR_PROCESS_LOGIC(108, "Wrong data processing logic"),
+ SPLIT_ATTR_ERROR(109, "Split attributes failure"),
UNSUPPORTED_MSG_TYPE(110, "Unsupported msgType"),
EMPTY_MSG(111, "Empty message"),
@@ -46,6 +58,8 @@ public enum DataProxyErrCode {
MQ_RETURN_ERROR(119, "MQ client return error"),
DUPLICATED_MESSAGE(120, "Duplicated message"),
+ GROUPID_OR_STREAMID_NOT_CONFIGURE(121, "GroupId or StreamId not found in
configure"),
+ GROUPID_OR_STREAMID_INCONSTANT(122, "GroupId or StreamId inconstant"),
UNKNOWN_ERROR(Integer.MAX_VALUE, "Unknown error");
@@ -78,4 +92,10 @@ public enum DataProxyErrCode {
public String getErrMsg() {
return errMsg;
}
+
+ public static String getErrMsg(String errCode) {
+ int codeVal = NumberUtils.toInt(errCode, Integer.MAX_VALUE);
+ return valueOf(codeVal).errMsg;
+ }
+
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/SinkRspEvent.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/SinkRspEvent.java
index ce79f0f4e..601f60e29 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/SinkRspEvent.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/SinkRspEvent.java
@@ -17,21 +17,21 @@
package org.apache.inlong.dataproxy.base;
-import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.Channel;
import java.util.Map;
import org.apache.flume.Event;
import org.apache.inlong.common.msg.MsgType;
public class SinkRspEvent implements Event {
- private ChannelHandlerContext ctx;
+ private Channel channel;
private MsgType msgType;
private Event event;
- public SinkRspEvent(Event event, MsgType msgType, ChannelHandlerContext
ctx) {
+ public SinkRspEvent(Event event, MsgType msgType, Channel channel) {
this.event = event;
this.msgType = msgType;
- this.ctx = ctx;
+ this.channel = channel;
}
@Override
@@ -59,8 +59,8 @@ public class SinkRspEvent implements Event {
*
* @return ctx
*/
- public ChannelHandlerContext getCtx() {
- return ctx;
+ public Channel getChannel() {
+ return channel;
}
/**
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/consts/StatConstants.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/consts/StatConstants.java
index b967e43f5..07b24c26b 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/consts/StatConstants.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/consts/StatConstants.java
@@ -27,11 +27,34 @@ public class StatConstants {
public static final java.lang.String METASINK_NOSLAVE = "metasink.noslave";
public static final java.lang.String METASINK_MSG_NOTOPIC =
"metasink.msgnotopic";
public static final java.lang.String METASINK_PROCESS_SPEED =
"metasink.process.speed";
- public static final java.lang.String EVENT_SUCCESS = "socketmsg.success";
- public static final java.lang.String EVENT_DROPPED = "socketmsg.dropped";
- public static final java.lang.String EVENT_EMPTY = "socketmsg.empty";
public static final java.lang.String EVENT_OTHEREXP = "socketmsg.otherexp";
public static final java.lang.String EVENT_INVALID = "socketmsg.invalid";
+ public static final java.lang.String EVENT_LINKS_OVERMAX = "links.overmax";
+ public static final java.lang.String EVENT_LINKS_ILLEGAL = "links.illegal";
+ public static final java.lang.String EVENT_LINKS_IN = "links.linkin";
+ public static final java.lang.String EVENT_LINKS_OUT = "links.linkout";
+ public static final java.lang.String EVENT_LINKS_EXCEPTION =
"links.exception";
+ public static final java.lang.String EVENT_EMPTY = "socketmsg.empty";
+ public static final java.lang.String EVENT_OVERMAXLEN =
"socketmsg.overmaxlen";
+ public static final java.lang.String EVENT_NOTEQUALLEN =
"socketmsg.notequallen";
+ public static final java.lang.String EVENT_MSGUNKNOWN_V0 =
"socketmsg.unknownV0";
+ public static final java.lang.String EVENT_MSGUNKNOWN_V1 =
"socketmsg.unknownV1";
+ public static final java.lang.String EVENT_MALFORMED =
"socketmsg.malformed";
+ public static final java.lang.String EVENT_NOBODY = "socketmsg.nobody";
+ public static final java.lang.String EVENT_NEGBODY = "socketmsg.negbody";
+ public static final java.lang.String EVENT_NEGATTR = "socketmsg.negattr";
+ public static final java.lang.String EVENT_INVALIDATTR =
"socketmsg.invattr";
+ public static final java.lang.String EVENT_UNSUPMSG = "socketmsg.unsupmsg";
+ public static final java.lang.String EVENT_UNPRESSEXP =
"socketmsg.upressexp";
+ public static final java.lang.String EVENT_WITHOUTGROUPID =
"socketmsg.wogroupid";
+ public static final java.lang.String EVENT_INCONSGROUPORSTREAMID =
"socketmsg.inconsids";
+ public static final java.lang.String EVENT_CHANNEL_NOT_WRITABLE =
"socketch.notwritable";
+ public static final java.lang.String EVENT_SERVICE_CLOSED =
"source.srvclosed";
+ public static final java.lang.String EVENT_SERVICE_UNREADY =
"sink.unready";
+ public static final java.lang.String EVENT_NOTOPIC = "config.notopic";
+ public static final java.lang.String EVENT_POST_SUCCESS =
"socketmsg.success";
+ public static final java.lang.String EVENT_POST_DROPPED =
"socketmsg.dropped";
+
public static final java.lang.String AGENT_MESSAGES_SENT_SUCCESS =
"agent.messages.success";
public static final java.lang.String AGENT_PACKAGES_SENT_SUCCESS =
"agent.packages.success";
public static final java.lang.String MSG_COUNTER_KEY = "msgcnt";
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/OrderBatchPackProfileV0.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/OrderBatchPackProfileV0.java
index 299a378b4..9faadf9dc 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/OrderBatchPackProfileV0.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/OrderBatchPackProfileV0.java
@@ -92,14 +92,14 @@ public class OrderBatchPackProfileV0 extends
BatchPackProfile {
}
return;
}
- if (orderProfile.getCtx() != null &&
orderProfile.getCtx().channel().isActive()) {
- orderProfile.getCtx().channel().eventLoop().execute(() -> {
+ if (orderProfile.getChannel() != null &&
orderProfile.getChannel().isActive()) {
+ orderProfile.getChannel().eventLoop().execute(() -> {
if (LOG.isDebugEnabled()) {
LOG.debug("order message rsp: seqId = {}, inlongGroupId =
{}, inlongStreamId = {}", sequenceId,
this.getInlongGroupId(), this.getInlongStreamId());
}
ByteBuf binBuffer = getResponsePackage("",
MsgType.MSG_BIN_MULTI_BODY, sequenceId);
- orderProfile.getCtx().writeAndFlush(binBuffer);
+ orderProfile.getChannel().writeAndFlush(binBuffer);
});
}
}
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 085dd6258..9f4d83d97 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
@@ -524,7 +524,7 @@ public class ServerMessageHandler extends
ChannelInboundHandlerAdapter {
Pair<Boolean, String> evenProcType =
MessageUtils.getEventProcType(syncSend, proxySend);
if (evenProcType.getLeft()) {
- event = new SinkRspEvent(event, msgType, ctx);
+ event = new SinkRspEvent(event, msgType, ctx.channel());
}
// build metric data item
long longDataTime = Long.parseLong(strDataTime);
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/BaseSource.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/BaseSource.java
new file mode 100644
index 000000000..6fd784681
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/BaseSource.java
@@ -0,0 +1,544 @@
+/*
+ * 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.dataproxy.source2;
+
+import com.google.common.base.Preconditions;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.util.concurrent.GlobalEventExecutor;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flume.ChannelSelector;
+import org.apache.flume.Context;
+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.common.metric.MetricRegister;
+import org.apache.inlong.common.monitor.MonitorIndex;
+import org.apache.inlong.common.monitor.MonitorIndexExt;
+import org.apache.inlong.dataproxy.admin.ProxyServiceMBean;
+import org.apache.inlong.dataproxy.channel.FailoverChannelProcessor;
+import org.apache.inlong.dataproxy.config.CommonConfigHolder;
+import org.apache.inlong.dataproxy.metrics.DataProxyMetricItemSet;
+import org.apache.inlong.dataproxy.utils.ConfStringUtils;
+import org.apache.inlong.dataproxy.utils.FailoverChannelProcessorHolder;
+import org.apache.inlong.sdk.commons.admin.AdminServiceRegister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.util.Map;
+
+/**
+ * source base class
+ *
+ */
+public abstract class BaseSource
+ extends
+ AbstractSource
+ implements
+ ProxyServiceMBean,
+ EventDrivenSource,
+ Configurable {
+
+ private static final Logger logger =
LoggerFactory.getLogger(BaseSource.class);
+
+ protected Context context;
+ // whether source reject service
+ protected volatile boolean isRejectService = false;
+ // source service host
+ protected String srcHost;
+ // source serviced port
+ protected int srcPort;
+ protected String strPort;
+ // message factory name
+ protected String msgFactoryName;
+ // message handler name
+ protected String messageHandlerName;
+ // source default topic
+ protected String defTopic = "";
+ // source default append attribute
+ protected String defAttr = "";
+ // allowed max message length
+ protected int maxMsgLength;
+ // whether compress message
+ protected boolean isCompressed;
+ // whether filter empty message
+ protected boolean filterEmptyMsg;
+ // whether custom channel processor
+ protected boolean customProcessor;
+ // max netty worker threads
+ protected int maxWorkerThreads;
+ // max netty accept threads
+ protected int maxAcceptThreads;
+ // max read idle time
+ protected long maxReadIdleTimeMs;
+ // max connection count
+ protected int maxConnections;
+ // netty parameters
+ protected EventLoopGroup acceptorGroup;
+ protected EventLoopGroup workerGroup;
+ protected ChannelGroup allChannels;
+ protected ChannelFuture channelFuture;
+ // receive buffer size
+ protected int maxRcvBufferSize;
+ // send buffer size
+ protected int maxSendBufferSize;
+ // file metric statistic
+ protected boolean fileMetricOn;
+ protected int monitorStatInvlSec;
+ protected int maxMonitorStatCnt;
+ protected MonitorIndex monitorIndex = null;
+ private MonitorIndexExt monitorIndexExt = null;
+ // metric set
+ protected DataProxyMetricItemSet metricItemSet;
+
+ public BaseSource() {
+ super();
+ allChannels = new DefaultChannelGroup("DefaultChannelGroup",
GlobalEventExecutor.INSTANCE);
+ }
+
+ @Override
+ public void configure(Context context) {
+ this.context = context;
+ this.srcHost = getHostIp(context);
+ this.srcPort = getHostPort(context);
+ this.strPort = String.valueOf(this.srcPort);
+ // get message factory
+ String tmpVal =
context.getString(SourceConstants.SRCCXT_MSG_FACTORY_NAME,
+ InLongMessageFactory.class.getName()).trim();
+ Preconditions.checkArgument(StringUtils.isNotBlank(tmpVal),
+ SourceConstants.SRCCXT_MSG_FACTORY_NAME + " config is blank");
+ this.msgFactoryName = tmpVal.trim();
+ // get message handler
+ tmpVal = context.getString(SourceConstants.SRCCXT_MESSAGE_HANDLER_NAME,
+ InLongMessageHandler.class.getName().trim());
+ Preconditions.checkArgument(StringUtils.isNotBlank(tmpVal),
+ SourceConstants.SRCCXT_MESSAGE_HANDLER_NAME + " config is
blank");
+ this.messageHandlerName = tmpVal;
+ // get default topic
+ tmpVal = context.getString(SourceConstants.SRCCXT_DEF_TOPIC);
+ if (StringUtils.isNotBlank(tmpVal)) {
+ this.defTopic = tmpVal.trim();
+ }
+ // get default attributes
+ tmpVal = context.getString(SourceConstants.SRCCXT_DEF_ATTR);
+ if (StringUtils.isNotBlank(tmpVal)) {
+ this.defAttr = tmpVal.trim();
+ }
+ // get allowed max message length
+ this.maxMsgLength = getIntValue(context,
SourceConstants.SRCCXT_MAX_MSG_LENGTH,
+ SourceConstants.VAL_DEF_MAX_MSG_LENGTH);
+ Preconditions.checkArgument((this.maxMsgLength >=
SourceConstants.VAL_MIN_MAX_MSG_LENGTH
+ && this.maxMsgLength <=
SourceConstants.VAL_MAX_MAX_MSG_LENGTH),
+ SourceConstants.SRCCXT_MAX_MSG_LENGTH + " must be in ["
+ + SourceConstants.VAL_MIN_MAX_MSG_LENGTH + ", "
+ + SourceConstants.VAL_MAX_MAX_MSG_LENGTH + "]");
+ // get whether compress message
+ this.isCompressed =
context.getBoolean(SourceConstants.SRCCXT_MSG_COMPRESSED,
+ SourceConstants.VAL_DEF_MSG_COMPRESSED);
+ // get whether filter empty message
+ this.filterEmptyMsg =
context.getBoolean(SourceConstants.SRCCXT_FILTER_EMPTY_MSG,
+ SourceConstants.VAL_DEF_FILTER_EMPTY_MSG);
+ // get whether custom channel processor
+ this.customProcessor =
context.getBoolean(SourceConstants.SRCCXT_CUSTOM_CHANNEL_PROCESSOR,
+ SourceConstants.VAL_DEF_CUSTOM_CH_PROCESSOR);
+ // get max accept threads
+ this.maxAcceptThreads = getIntValue(context,
SourceConstants.SRCCXT_MAX_ACCEPT_THREADS,
+ SourceConstants.VAL_DEF_NET_ACCEPT_THREADS);
+ Preconditions.checkArgument((this.maxAcceptThreads >=
SourceConstants.VAL_MIN_ACCEPT_THREADS
+ && this.maxAcceptThreads <=
SourceConstants.VAL_MAX_ACCEPT_THREADS),
+ SourceConstants.SRCCXT_MAX_ACCEPT_THREADS + " must be in ["
+ + SourceConstants.VAL_MIN_ACCEPT_THREADS + ", "
+ + SourceConstants.VAL_MAX_ACCEPT_THREADS + "]");
+ // get max worker threads
+ this.maxWorkerThreads = getIntValue(context,
SourceConstants.SRCCXT_MAX_WORKER_THREADS,
+ SourceConstants.VAL_DEF_WORKER_THREADS);
+ Preconditions.checkArgument((this.maxWorkerThreads >=
SourceConstants.VAL_MIN_WORKER_THREADS
+ && this.maxWorkerThreads <=
SourceConstants.VAL_MAX_WORKER_THREADS),
+ SourceConstants.SRCCXT_MAX_WORKER_THREADS + " must be in ["
+ + SourceConstants.VAL_MIN_WORKER_THREADS + ", "
+ + SourceConstants.VAL_MAX_WORKER_THREADS + "]");
+ // get max read idle time
+ this.maxReadIdleTimeMs = getLongValue(context,
SourceConstants.SRCCXT_MAX_READ_IDLE_TIME_MS,
+ SourceConstants.VAL_DEF_READ_IDLE_TIME_MS);
+ Preconditions.checkArgument((this.maxReadIdleTimeMs >=
SourceConstants.VAL_MIN_READ_IDLE_TIME_MS),
+ SourceConstants.SRCCXT_MAX_READ_IDLE_TIME_MS + " must be >= "
+ + SourceConstants.VAL_MIN_READ_IDLE_TIME_MS);
+ // get file metric statistic
+ this.monitorStatInvlSec = getIntValue(context,
SourceConstants.SRCCXT_STAT_INTERVAL_SEC,
+ SourceConstants.VAL_DEF_STAT_INVL_SEC);
+ Preconditions.checkArgument((this.monitorStatInvlSec >=
SourceConstants.VAL_MIN_STAT_INVL_SEC),
+ SourceConstants.SRCCXT_STAT_INTERVAL_SEC + " must be >= "
+ + SourceConstants.VAL_MIN_STAT_INVL_SEC);
+ // get max monitor key count
+ this.maxMonitorStatCnt = getIntValue(context,
SourceConstants.SRCCXT_MAX_MONITOR_STAT_CNT,
+ SourceConstants.VAL_DEF_MON_STAT_CNT);
+ Preconditions.checkArgument(this.maxMonitorStatCnt >=
SourceConstants.VAL_MIN_MON_STAT_CNT,
+ SourceConstants.SRCCXT_MAX_MONITOR_STAT_CNT + " must be >= "
+ + SourceConstants.VAL_MIN_MON_STAT_CNT);
+ // get max connect count
+ this.maxConnections = getIntValue(context,
SourceConstants.SRCCXT_MAX_CONNECTION_CNT,
+ SourceConstants.VAL_DEF_MAX_CONNECTION_CNT);
+ Preconditions.checkArgument(this.maxConnections >=
SourceConstants.VAL_MIN_CONNECTION_CNT,
+ SourceConstants.SRCCXT_MAX_CONNECTION_CNT + " must be >= "
+ + SourceConstants.VAL_MIN_CONNECTION_CNT);
+ // get whether enable file metric
+ this.fileMetricOn =
context.getBoolean(SourceConstants.SRCCXT_FILE_METRIC_ON,
+ SourceConstants.VAL_DEF_FILE_METRIC_ON);
+ // get max receive buffer size
+ this.maxRcvBufferSize = getIntValue(context,
SourceConstants.SRCCXT_RECEIVE_BUFFER_SIZE,
+ SourceConstants.VAL_DEF_RECEIVE_BUFFER_SIZE);
+ Preconditions.checkArgument(this.maxRcvBufferSize >=
SourceConstants.VAL_MIN_RECEIVE_BUFFER_SIZE,
+ SourceConstants.SRCCXT_RECEIVE_BUFFER_SIZE + " must be >= "
+ + SourceConstants.VAL_MIN_RECEIVE_BUFFER_SIZE);
+ if (this.maxRcvBufferSize >
SourceConstants.VAL_MAX_RECEIVE_BUFFER_SIZE) {
+ this.maxRcvBufferSize =
SourceConstants.VAL_MAX_RECEIVE_BUFFER_SIZE;
+ }
+ // get max send buffer size
+ this.maxSendBufferSize = getIntValue(context,
SourceConstants.SRCCXT_SEND_BUFFER_SIZE,
+ SourceConstants.VAL_DEF_SEND_BUFFER_SIZE);
+ Preconditions.checkArgument(this.maxSendBufferSize >=
SourceConstants.VAL_MIN_SEND_BUFFER_SIZE,
+ SourceConstants.SRCCXT_SEND_BUFFER_SIZE + " must be >= "
+ + SourceConstants.VAL_MIN_SEND_BUFFER_SIZE);
+ if (this.maxSendBufferSize > SourceConstants.VAL_MAX_SEND_BUFFER_SIZE)
{
+ this.maxSendBufferSize = SourceConstants.VAL_MAX_SEND_BUFFER_SIZE;
+ }
+ }
+
+ @Override
+ public synchronized void start() {
+ if (customProcessor) {
+ ChannelSelector selector = getChannelProcessor().getSelector();
+ FailoverChannelProcessor newProcessor = new
FailoverChannelProcessor(selector);
+ newProcessor.configure(this.context);
+ setChannelProcessor(newProcessor);
+ FailoverChannelProcessorHolder.setChannelProcessor(newProcessor);
+ }
+ super.start();
+ // initial metric item set
+ this.metricItemSet = new DataProxyMetricItemSet(
+ CommonConfigHolder.getInstance().getClusterName(), getName(),
String.valueOf(srcPort));
+ MetricRegister.register(metricItemSet);
+ // init monitor logic
+ if (fileMetricOn) {
+ this.monitorIndex = new MonitorIndex("Source", monitorStatInvlSec,
maxMonitorStatCnt);
+ this.monitorIndexExt = new MonitorIndexExt(
+ "DataProxy_monitors#" + this.getProtocolName(),
monitorStatInvlSec, maxMonitorStatCnt);
+ }
+ startSource();
+ // register
+ AdminServiceRegister.register(ProxyServiceMBean.MBEAN_TYPE,
this.getName(), this);
+ }
+
+ @Override
+ public synchronized void stop() {
+ logger.info("[STOP {} SOURCE]{} stopping...", this.getProtocolName(),
this.getName());
+ // close channels
+ if (!allChannels.isEmpty()) {
+ try {
+ allChannels.close().awaitUninterruptibly();
+ } catch (Exception e) {
+ logger.warn("Close {} netty channels throw exception",
this.getName(), e);
+ } finally {
+ allChannels.clear();
+ }
+ }
+ // close channel future
+ if (channelFuture != null) {
+ try {
+ channelFuture.channel().closeFuture().sync();
+ } catch (InterruptedException e) {
+ logger.warn("Close {} channel future throw exception",
this.getName(), e);
+ }
+ }
+ // stop super class
+ super.stop();
+ // stop file statistic index
+ if (fileMetricOn) {
+ if (monitorIndex != null) {
+ monitorIndex.shutDown();
+ }
+ if (monitorIndexExt != null) {
+ monitorIndexExt.shutDown();
+ }
+ }
+ logger.info("[STOP {} SOURCE]{} stopped", this.getProtocolName(),
this.getName());
+ }
+
+ /**
+ * get metricItemSet
+ * @return the metricItemSet
+ */
+ public DataProxyMetricItemSet getMetricItemSet() {
+ return metricItemSet;
+ }
+
+ public Context getContext() {
+ return context;
+ }
+
+ public String getSrcHost() {
+ return srcHost;
+ }
+
+ public int getSrcPort() {
+ return srcPort;
+ }
+
+ public String getStrPort() {
+ return strPort;
+ }
+
+ public String getDefTopic() {
+ return defTopic;
+ }
+
+ public String getDefAttr() {
+ return defAttr;
+ }
+
+ public int getMaxMsgLength() {
+ return maxMsgLength;
+ }
+
+ public boolean isCompressed() {
+ return isCompressed;
+ }
+
+ public boolean isFilterEmptyMsg() {
+ return filterEmptyMsg;
+ }
+
+ public boolean isCustomProcessor() {
+ return customProcessor;
+ }
+
+ public int getMaxConnections() {
+ return maxConnections;
+ }
+
+ public ChannelGroup getAllChannels() {
+ return allChannels;
+ }
+
+ public long getMaxReadIdleTimeMs() {
+ return maxReadIdleTimeMs;
+ }
+
+ public String getMessageHandlerName() {
+ return messageHandlerName;
+ }
+
+ public int getMaxWorkerThreads() {
+ return maxWorkerThreads;
+ }
+
+ public void fileMetricEventInc(String eventKey) {
+ if (fileMetricOn) {
+ monitorIndexExt.incrementAndGet(eventKey);
+ }
+ }
+
+ public void fileMetricRecordAdd(String key, int cnt, int packCnt, long
packSize, int failCnt) {
+ if (fileMetricOn) {
+ monitorIndex.addAndGet(key, cnt, packCnt, packSize, failCnt);
+ }
+ }
+
+ /**
+ * channel factory
+ * @return
+ */
+ public ChannelInitializer getChannelInitializerFactory() {
+ ChannelInitializer fac = null;
+ logger.info(this.getName() + " load msgFactory=" + msgFactoryName);
+ try {
+ Class<? extends ChannelInitializer> clazz =
+ (Class<? extends ChannelInitializer>)
Class.forName(msgFactoryName);
+ Constructor ctor = clazz.getConstructor(BaseSource.class);
+ logger.info("Using channel processor:{}",
getChannelProcessor().getClass().getName());
+ fac = (ChannelInitializer) ctor.newInstance(this);
+ } catch (Exception e) {
+ logger.error("{} start error, fail to construct
ChannelPipelineFactory with name {}",
+ this.getName(), msgFactoryName, e);
+ stop();
+ throw new FlumeException(e.getMessage());
+ }
+ return fac;
+ }
+
+ public abstract String getProtocolName();
+
+ public abstract void startSource();
+
+ /**
+ * stopService
+ */
+ @Override
+ public void stopService() {
+ this.isRejectService = true;
+ }
+
+ /**
+ * recoverService
+ */
+ @Override
+ public void recoverService() {
+ this.isRejectService = false;
+ }
+
+ /**
+ * isRejectService
+ *
+ * @return
+ */
+ public boolean isRejectService() {
+ return isRejectService;
+ }
+
+ /**
+ * Get the configuration value of integer type from the context
+ *
+ * @param context the context
+ * @param fieldKey the configure key
+ * @param defVal the default value
+ *
+ * @return the configuration value
+ */
+ public int getIntValue(Context context, String fieldKey, int defVal) {
+ String tmpVal = context.getString(fieldKey);
+ if (StringUtils.isNotBlank(tmpVal)) {
+ int result;
+ tmpVal = tmpVal.trim();
+ try {
+ result = Integer.parseInt(tmpVal);
+ } catch (Throwable e) {
+ throw new IllegalArgumentException(
+ fieldKey + "(" + tmpVal + ") must specify an integer
value!");
+ }
+ return result;
+ }
+ return defVal;
+ }
+
+ /**
+ * Get the configuration value of long type from the context
+ *
+ * @param context the context
+ * @param fieldKey the configure key
+ * @param defVal the default value
+ *
+ * @return the configuration value
+ */
+ public long getLongValue(Context context, String fieldKey, long defVal) {
+ String tmpVal = context.getString(fieldKey);
+ if (StringUtils.isNotBlank(tmpVal)) {
+ long result;
+ tmpVal = tmpVal.trim();
+ try {
+ result = Long.parseLong(tmpVal);
+ } catch (Throwable e) {
+ throw new IllegalArgumentException(
+ fieldKey + "(" + tmpVal + ") must specify an long
value!");
+ }
+ return result;
+ }
+ return defVal;
+ }
+
+ /**
+ * getHostIp
+ *
+ * @param context
+ * @return
+ */
+ private String getHostIp(Context context) {
+ String result = null;
+ // first get host ip from dataProxy.conf
+ String tmpVal = context.getString(SourceConstants.SRCCXT_CONFIG_HOST);
+ if (StringUtils.isNotBlank(tmpVal)) {
+ tmpVal = tmpVal.trim();
+ Preconditions.checkArgument(ConfStringUtils.isValidIp(tmpVal),
+ SourceConstants.SRCCXT_CONFIG_HOST + "(" + tmpVal + ")
config in conf not valid");
+ result = tmpVal;
+ }
+ // second get host ip from system env
+ Map<String, String> envMap = System.getenv();
+ if (envMap.containsKey(SourceConstants.SYSENV_HOST_IP)) {
+ tmpVal = envMap.get(SourceConstants.SYSENV_HOST_IP);
+ Preconditions.checkArgument(ConfStringUtils.isValidIp(tmpVal),
+ SourceConstants.SYSENV_HOST_IP + "(" + tmpVal + ") config
in system env not valid");
+ result = tmpVal.trim();
+ }
+ if (StringUtils.isBlank(result)) {
+ result = SourceConstants.VAL_DEF_HOST_VALUE;
+ }
+ return result;
+ }
+
+ /**
+ * getHostPort
+ *
+ * @param context
+ * @return
+ */
+ private int getHostPort(Context context) {
+ Integer result = null;
+ // first get host port from dataProxy.conf
+ String tmpVal = context.getString(SourceConstants.SRCCXT_CONFIG_PORT);
+ if (StringUtils.isNotBlank(tmpVal)) {
+ tmpVal = tmpVal.trim();
+ try {
+ result = Integer.parseInt(tmpVal);
+ } catch (Throwable e) {
+ throw new IllegalArgumentException(
+ SourceConstants.SYSENV_HOST_PORT + "(" + tmpVal + ")
config in conf not integer");
+ }
+ }
+ if (result != null) {
+ Preconditions.checkArgument(ConfStringUtils.isValidPort(result),
+ SourceConstants.SRCCXT_CONFIG_PORT + "(" + result + ")
config in conf not valid");
+ }
+ // second get host port from system env
+ Map<String, String> envMap = System.getenv();
+ if (envMap.containsKey(SourceConstants.SYSENV_HOST_PORT)) {
+ tmpVal = envMap.get(SourceConstants.SYSENV_HOST_PORT);
+ if (StringUtils.isNotBlank(tmpVal)) {
+ tmpVal = tmpVal.trim();
+ try {
+ result = Integer.parseInt(tmpVal);
+ } catch (Throwable e) {
+ throw new IllegalArgumentException(
+ SourceConstants.SYSENV_HOST_PORT + "(" + tmpVal +
") config in system env not integer");
+ }
+
Preconditions.checkArgument(ConfStringUtils.isValidPort(result),
+ SourceConstants.SYSENV_HOST_PORT + "(" + tmpVal + ")
config in system env not valid");
+ }
+ }
+ if (result == null) {
+ throw new IllegalArgumentException("Required parameter " +
+ SourceConstants.SRCCXT_CONFIG_PORT + " must exist and may
not be null");
+ }
+ return result;
+ }
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageFactory.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageFactory.java
new file mode 100644
index 000000000..2dfe66df8
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageFactory.java
@@ -0,0 +1,77 @@
+/*
+ * 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.dataproxy.source2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.lang.reflect.Constructor;
+import java.util.concurrent.TimeUnit;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+
+public class InLongMessageFactory extends ChannelInitializer<SocketChannel> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(InLongMessageFactory.class);
+
+ public static final int INLONG_LENGTH_FIELD_OFFSET = 0;
+ public static final int INLONG_LENGTH_FIELD_LENGTH = 4;
+ public static final int INLONG_LENGTH_ADJUSTMENT = -4;
+ public static final int INLONG_INITIAL_BYTES_TO_STRIP = 0;
+ public static final boolean DEFAULT_FAIL_FAST = true;
+
+ private BaseSource source;
+
+ /**
+ * get server factory
+ *
+ * @param source
+ */
+ public InLongMessageFactory(BaseSource source) {
+ this.source = source;
+ }
+
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+
+ if (source.getProtocolName()
+ .equalsIgnoreCase(SourceConstants.SRC_PROTOCOL_TYPE_TCP)) {
+ ch.pipeline().addLast("messageDecoder", new
LengthFieldBasedFrameDecoder(
+ source.getMaxMsgLength(), INLONG_LENGTH_FIELD_OFFSET,
INLONG_LENGTH_FIELD_LENGTH,
+ INLONG_LENGTH_ADJUSTMENT, INLONG_INITIAL_BYTES_TO_STRIP,
DEFAULT_FAIL_FAST));
+ ch.pipeline().addLast("readTimeoutHandler",
+ new ReadTimeoutHandler(source.getMaxReadIdleTimeMs(),
TimeUnit.MILLISECONDS));
+ }
+ // build message handler
+ if (source.getChannelProcessor() != null) {
+ try {
+ Class<? extends ChannelInboundHandlerAdapter> clazz =
+ (Class<? extends ChannelInboundHandlerAdapter>)
Class.forName(source.getMessageHandlerName());
+ Constructor<?> ctor = clazz.getConstructor(BaseSource.class);
+ ChannelInboundHandlerAdapter messageHandler =
+ (ChannelInboundHandlerAdapter)
ctor.newInstance(source);
+ ch.pipeline().addLast("messageHandler", messageHandler);
+ } catch (Exception e) {
+ LOG.error("{} newInstance {} failure!", source.getName(),
+ source.getMessageHandlerName(), e);
+ }
+ }
+ }
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageHandler.java
new file mode 100644
index 000000000..e92a58b98
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageHandler.java
@@ -0,0 +1,681 @@
+/*
+ * 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.dataproxy.source2;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flume.ChannelException;
+import org.apache.flume.Event;
+import org.apache.inlong.common.enums.DataProxyErrCode;
+import org.apache.inlong.common.monitor.LogCounter;
+import org.apache.inlong.common.msg.AttributeConstants;
+import org.apache.inlong.common.msg.MsgType;
+import org.apache.inlong.dataproxy.config.CommonConfigHolder;
+import org.apache.inlong.dataproxy.config.ConfigManager;
+import org.apache.inlong.dataproxy.consts.AttrConstants;
+import org.apache.inlong.dataproxy.consts.ConfigConstants;
+import org.apache.inlong.dataproxy.consts.StatConstants;
+import org.apache.inlong.dataproxy.metrics.DataProxyMetricItem;
+import org.apache.inlong.dataproxy.metrics.audit.AuditUtils;
+import org.apache.inlong.dataproxy.source.tcp.InlongTcpSourceCallback;
+import org.apache.inlong.dataproxy.source2.v0msg.AbsV0MsgCodec;
+import org.apache.inlong.dataproxy.source2.v0msg.CodecBinMsg;
+import org.apache.inlong.dataproxy.source2.v0msg.CodecTextMsg;
+import org.apache.inlong.dataproxy.utils.AddressUtils;
+import org.apache.inlong.dataproxy.utils.DateTimeUtils;
+import org.apache.inlong.sdk.commons.protocol.EventUtils;
+import org.apache.inlong.sdk.commons.protocol.ProxyEvent;
+import org.apache.inlong.sdk.commons.protocol.ProxyPackEvent;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import static
org.apache.inlong.dataproxy.source2.InLongMessageFactory.INLONG_LENGTH_FIELD_LENGTH;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_HB_ATTRLEN_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_HB_BODYLEN_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_HB_BODY_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_HB_FIXED_CONTENT_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_HB_FORMAT_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_HB_TOTALLEN_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_HB_VERSION_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_FIXED_CONTENT_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_MAGIC;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.TXT_MSG_FIXED_CONTENT_SIZE;
+
+/**
+ * Server message handler
+ *
+ */
+public class InLongMessageHandler extends ChannelInboundHandlerAdapter {
+
+ private static final Logger logger =
LoggerFactory.getLogger(InLongMessageHandler.class);
+ // log print count
+ private static final LogCounter logCounter = new LogCounter(10, 100000, 30
* 1000);
+
+ private static final int INLONG_MSG_V1 = 1;
+ private static final String DEFAULT_REMOTE_IDC_VALUE = "0";
+
+ private static final ConfigManager configManager =
ConfigManager.getInstance();
+ private final BaseSource source;
+
+ /**
+ * Constructor
+ *
+ * @param source AbstractSource
+ */
+ public InLongMessageHandler(BaseSource source) {
+ this.source = source;
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws
Exception {
+ if (msg == null) {
+ source.fileMetricEventInc(StatConstants.EVENT_EMPTY);
+ logger.debug("Get null msg, just skip!");
+ return;
+ }
+ ByteBuf cb = (ByteBuf) msg;
+ try {
+ int readableLength = cb.readableBytes();
+ if (readableLength == 0 && source.isFilterEmptyMsg()) {
+ cb.clear();
+ source.fileMetricEventInc(StatConstants.EVENT_EMPTY);
+ logger.debug("skip empty msg.");
+ return;
+ }
+ if (readableLength > source.getMaxMsgLength()) {
+ source.fileMetricEventInc(StatConstants.EVENT_OVERMAXLEN);
+ throw new Exception("Error msg, readableLength(" +
readableLength +
+ ") > max allowed message length (" +
source.getMaxMsgLength() + ")");
+ }
+ // save index
+ cb.markReaderIndex();
+ // read total data length
+ int totalDataLen = cb.readInt();
+ if (readableLength < totalDataLen + INLONG_LENGTH_FIELD_LENGTH) {
+ // reset index when buffer is not satisfied.
+ cb.resetReaderIndex();
+ source.fileMetricEventInc(StatConstants.EVENT_NOTEQUALLEN);
+ throw new Exception("Error msg, channel buffer is not
satisfied, and readableLength="
+ + readableLength + ", and totalPackLength=" +
totalDataLen + " + 4");
+ }
+ // read type
+ int msgTypeValue = cb.readByte();
+ if (msgTypeValue == 0x0) {
+ // process v1 messsages
+ msgTypeValue = cb.readByte();
+ if (msgTypeValue == INLONG_MSG_V1) {
+ // decode version 1
+ int bodyLength = totalDataLen - 2;
+ processV1Msg(ctx, cb, bodyLength);
+ } else {
+ // unknown message type
+
source.fileMetricEventInc(StatConstants.EVENT_MSGUNKNOWN_V1);
+ throw new Exception("Unknown V1 message version, version =
" + msgTypeValue);
+ }
+ } else {
+ // process v0 messages
+ Channel channel = ctx.channel();
+ MsgType msgType = MsgType.valueOf(msgTypeValue);
+ final long msgRcvTime = System.currentTimeMillis();
+ if (MsgType.MSG_UNKNOWN == msgType) {
+
source.fileMetricEventInc(StatConstants.EVENT_MSGUNKNOWN_V0);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Received unknown message, channel {}",
channel);
+ }
+ throw new Exception("Unknown V0 message type, type = " +
msgTypeValue);
+ } else if (MsgType.MSG_HEARTBEAT == msgType) {
+ // send response message
+ flushV0MsgPackage(source, channel,
buildHeartBeatMsgRspPackage(), MsgType.MSG_HEARTBEAT.name());
+ return;
+ } else if (MsgType.MSG_BIN_HEARTBEAT == msgType) {
+ procBinHeartbeatMsg(source, channel, cb, totalDataLen);
+ return;
+ }
+ // process msgType in {2,3,4,5,6,7}
+ AbsV0MsgCodec msgCodec;
+ String strRemoteIP = AddressUtils.getChannelRemoteIP(channel);
+ // check whether totalDataLen is valid.
+ if (MsgType.MSG_BIN_MULTI_BODY == msgType) {
+ if (totalDataLen < BIN_MSG_FIXED_CONTENT_SIZE) {
+
source.fileMetricEventInc(StatConstants.EVENT_MALFORMED);
+ String errMsg = String.format("Malformed msg,
totalDataLen(%d) < min bin7-msg length(%d)",
+ totalDataLen, BIN_MSG_FIXED_CONTENT_SIZE);
+ if (logger.isDebugEnabled()) {
+ logger.debug(errMsg + ", channel {}", channel);
+ }
+ throw new Exception(errMsg);
+ }
+ msgCodec = new CodecBinMsg(totalDataLen, msgTypeValue,
msgRcvTime, strRemoteIP);
+ } else {
+ if (totalDataLen < TXT_MSG_FIXED_CONTENT_SIZE) {
+
source.fileMetricEventInc(StatConstants.EVENT_MALFORMED);
+ String errMsg = String.format("Malformed msg,
totalDataLen(%d) < min txt-msg length(%d)",
+ totalDataLen, TXT_MSG_FIXED_CONTENT_SIZE);
+ if (logger.isDebugEnabled()) {
+ logger.debug(errMsg + ", channel {}", channel);
+ }
+ throw new Exception(errMsg);
+ }
+ msgCodec = new CodecTextMsg(totalDataLen, msgTypeValue,
msgRcvTime, strRemoteIP);
+ }
+ // process request
+ processV0Msg(channel, cb, msgCodec);
+ }
+ } finally {
+ cb.release();
+ }
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ // check max allowed connection count
+ if (source.getAllChannels().size() >= source.getMaxConnections()) {
+ source.fileMetricEventInc(StatConstants.EVENT_LINKS_OVERMAX);
+ ctx.channel().disconnect();
+ ctx.channel().close();
+ logger.warn("{} refuse to connect = {} , connections = {},
maxConnections = {}",
+ source.getName(), ctx.channel(),
source.getAllChannels().size(), source.getMaxConnections());
+ return;
+ }
+ // check illegal ip
+ if (ConfigManager.getInstance().needChkIllegalIP()) {
+ String strRemoteIp =
AddressUtils.getChannelRemoteIP(ctx.channel());
+ if (strRemoteIp != null
+ && ConfigManager.getInstance().isIllegalIP(strRemoteIp)) {
+ source.fileMetricEventInc(StatConstants.EVENT_LINKS_ILLEGAL);
+ ctx.channel().disconnect();
+ ctx.channel().close();
+ logger.error(strRemoteIp + " is Illegal IP, so refuse it !");
+ return;
+ }
+ }
+ // add legal channel
+ source.getAllChannels().add(ctx.channel());
+ ctx.fireChannelActive();
+ source.fileMetricEventInc(StatConstants.EVENT_LINKS_IN);
+ logger.info("{} added new channel {}, current connections = {},
maxConnections = {}",
+ source.getName(), ctx.channel(),
source.getAllChannels().size(), source.getMaxConnections());
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) {
+ logger.error("{} channel {} inactive", source.getName(),
ctx.channel());
+ ctx.fireChannelInactive();
+ source.getAllChannels().remove(ctx.channel());
+ source.fileMetricEventInc(StatConstants.EVENT_LINKS_OUT);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
+ logger.error("{} channel {} throw exception", source.getName(),
ctx.channel(), cause);
+ ctx.fireExceptionCaught(cause);
+ if (ctx.channel() != null) {
+ try {
+ ctx.channel().disconnect();
+ ctx.channel().close();
+ } catch (Exception ex) {
+ //
+ }
+ source.getAllChannels().remove(ctx.channel());
+ source.fileMetricEventInc(StatConstants.EVENT_LINKS_EXCEPTION);
+ }
+ ctx.close();
+ }
+
+ private void processV0Msg(Channel channel, ByteBuf cb, AbsV0MsgCodec
msgCodec) throws Exception {
+ final StringBuilder strBuff = new StringBuilder(512);
+ // decode the request message
+ if (!msgCodec.descMsg(source, cb)) {
+ responseV0Msg(channel, msgCodec, strBuff);
+ return;
+ }
+ // check service status.
+ if (source.isRejectService()) {
+ source.fileMetricEventInc(StatConstants.EVENT_SERVICE_CLOSED);
+ msgCodec.setFailureInfo(DataProxyErrCode.SERVICE_CLOSED);
+ responseV0Msg(channel, msgCodec, strBuff);
+ return;
+ }
+ // check if the node is linked to the Manager.
+ if (!ConfigManager.getInstance().isMqClusterReady()) {
+ source.fileMetricEventInc(StatConstants.EVENT_SERVICE_UNREADY);
+ msgCodec.setFailureInfo(DataProxyErrCode.SINK_SERVICE_UNREADY);
+ responseV0Msg(channel, msgCodec, strBuff);
+ return;
+ }
+ // valid and fill extra fields.
+ if (!msgCodec.validAndFillFields(source, strBuff)) {
+ responseV0Msg(channel, msgCodec, strBuff);
+ return;
+ }
+ // build InLong event.
+ Event event = msgCodec.encEventPackage(source, channel);
+ // build metric data item
+ long longDataTime = msgCodec.getDataTimeMs() / 1000 / 60 / 10;
+ longDataTime = longDataTime * 1000 * 60 * 10;
+
strBuff.append(source.getProtocolName()).append(AttrConstants.SEPARATOR)
+
.append(msgCodec.getTopicName()).append(AttrConstants.SEPARATOR)
+ .append(msgCodec.getStreamId()).append(AttrConstants.SEPARATOR)
+
.append(msgCodec.getStrRemoteIP()).append(AttrConstants.SEPARATOR)
+ .append(source.getStrPort()).append(AttrConstants.SEPARATOR)
+
.append(msgCodec.getMsgProcType()).append(AttrConstants.SEPARATOR)
+ .append(DateTimeUtils.ms2yyyyMMddHHmm(longDataTime))
+ .append(AttrConstants.SEPARATOR).append(
+
DateTimeUtils.ms2yyyyMMddHHmm(msgCodec.getMsgRcvTime()));
+ try {
+ source.getChannelProcessor().processEvent(event);
+ source.fileMetricEventInc(StatConstants.EVENT_POST_SUCCESS);
+ source.fileMetricRecordAdd(strBuff.toString(),
+ msgCodec.getMsgCount(), 1, msgCodec.getBodyLength(), 0);
+ this.addMetric(true, event.getBody().length, event);
+ strBuff.delete(0, strBuff.length());
+ } catch (Throwable ex) {
+ logger.error("Error writting to channel, data will discard.", ex);
+ source.fileMetricEventInc(StatConstants.EVENT_POST_DROPPED);
+ source.fileMetricRecordAdd(strBuff.toString(), 0, 0, 0,
msgCodec.getMsgCount());
+ this.addMetric(false, event.getBody().length, event);
+ strBuff.delete(0, strBuff.length());
+ throw new ChannelException("ProcessEvent error can't write event
to channel.");
+ }
+ }
+
+ private void processV1Msg(ChannelHandlerContext ctx, ByteBuf cb, int
bodyLength) throws Exception {
+ // read bytes
+ byte[] msgBytes = new byte[bodyLength];
+ cb.readBytes(msgBytes);
+ // decode
+ ProxySdk.MessagePack packObject =
ProxySdk.MessagePack.parseFrom(msgBytes);
+ // reject service
+ if (source.isRejectService()) {
+ this.addMetric(false, 0, null);
+ source.fileMetricEventInc(StatConstants.EVENT_SERVICE_CLOSED);
+ this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT,
packObject);
+ return;
+ }
+ // uncompress
+ List<ProxyEvent> events = EventUtils.decodeSdkPack(packObject);
+ // response success if event size is zero
+ if (events.size() == 0) {
+ this.responsePackage(ctx, ProxySdk.ResultCode.SUCCUSS, packObject);
+ }
+ // process
+ if (CommonConfigHolder.getInstance().isResponseAfterSave()) {
+ this.processAndWaitingSave(ctx, packObject, events);
+ } else {
+ this.processAndResponse(ctx, packObject, events);
+ }
+ }
+
+ /**
+ * responsePackage
+ *
+ * @param ctx
+ * @param code
+ * @throws Exception
+ */
+ private void responsePackage(ChannelHandlerContext ctx,
+ ProxySdk.ResultCode code,
+ ProxySdk.MessagePack packObject) throws Exception {
+ ProxySdk.ResponseInfo.Builder builder =
ProxySdk.ResponseInfo.newBuilder();
+ builder.setResult(code);
+ ProxySdk.MessagePackHeader header = packObject.getHeader();
+ builder.setPackId(header.getPackId());
+
+ // encode
+ byte[] responseBytes = builder.build().toByteArray();
+ //
+ ByteBuf buffer = Unpooled.wrappedBuffer(responseBytes);
+ Channel remoteChannel = ctx.channel();
+ if (remoteChannel.isWritable()) {
+ remoteChannel.write(buffer);
+ } else {
+ buffer.release();
+ logger.warn("Send buffer2 is not writable, disconnect {}",
remoteChannel);
+ throw new Exception("Send buffer2 is not writable, disconnect " +
remoteChannel);
+ }
+ }
+
+ /**
+ * processAndWaitingSave
+ * @param ctx
+ * @param packObject
+ * @param events
+ * @throws Exception
+ */
+ private void processAndWaitingSave(ChannelHandlerContext ctx,
+ ProxySdk.MessagePack packObject,
+ List<ProxyEvent> events) throws Exception {
+ ProxySdk.MessagePackHeader header = packObject.getHeader();
+ InlongTcpSourceCallback callback = new InlongTcpSourceCallback(ctx,
header);
+ String inlongGroupId = header.getInlongGroupId();
+ String inlongStreamId = header.getInlongStreamId();
+ ProxyPackEvent packEvent = new ProxyPackEvent(inlongGroupId,
inlongStreamId, events, callback);
+ // put to channel
+ try {
+ source.getChannelProcessor().processEvent(packEvent);
+ events.forEach(event -> {
+ this.addMetric(true, event.getBody().length, event);
+ source.fileMetricEventInc(StatConstants.EVENT_POST_SUCCESS);
+ });
+ boolean awaitResult = callback.getLatch().await(
+
CommonConfigHolder.getInstance().getMaxResAfterSaveTimeout(),
TimeUnit.MILLISECONDS);
+ if (!awaitResult) {
+ if (!callback.getHasResponsed().getAndSet(true)) {
+ this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT,
packObject);
+ }
+ }
+ } catch (Throwable ex) {
+ logger.error("Process Controller Event error can't write event to
channel.", ex);
+ events.forEach(event -> {
+ this.addMetric(false, event.getBody().length, event);
+ source.fileMetricEventInc(StatConstants.EVENT_POST_DROPPED);
+ });
+ if (!callback.getHasResponsed().getAndSet(true)) {
+ this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT,
packObject);
+ }
+ }
+ }
+
+ /**
+ * processAndResponse
+ * @param ctx
+ * @param packObject
+ * @param events
+ * @throws Exception
+ */
+ private void processAndResponse(ChannelHandlerContext ctx,
+ ProxySdk.MessagePack packObject,
+ List<ProxyEvent> events) throws Exception {
+ for (ProxyEvent event : events) {
+ // get configured topic name
+ String topic =
configManager.getTopicName(event.getInlongGroupId(), event.getInlongStreamId());
+ if (StringUtils.isBlank(topic)) {
+ if (CommonConfigHolder.getInstance().isNoTopicAccept()) {
+ topic = source.getDefTopic();
+ } else {
+ source.fileMetricEventInc(StatConstants.EVENT_NOTOPIC);
+ this.addMetric(false, event.getBody().length, event);
+ this.responsePackage(ctx,
ProxySdk.ResultCode.ERR_ID_ERROR, packObject);
+ return;
+ }
+ }
+ event.setTopic(topic);
+ // put to channel
+ try {
+ source.getChannelProcessor().processEvent(event);
+ this.addMetric(true, event.getBody().length, event);
+ source.fileMetricEventInc(StatConstants.EVENT_POST_SUCCESS);
+ } catch (Throwable ex) {
+ logger.error("Process Controller Event error can't write event
to channel.", ex);
+ this.addMetric(false, event.getBody().length, event);
+ this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT,
packObject);
+ source.fileMetricEventInc(StatConstants.EVENT_POST_DROPPED);
+ return;
+ }
+ }
+ this.responsePackage(ctx, ProxySdk.ResultCode.SUCCUSS, packObject);
+ }
+
+ /**
+ * Return response to client in source
+ */
+ private void responseV0Msg(Channel channel, AbsV0MsgCodec msgObj,
StringBuilder strBuff) throws Exception {
+ // check channel status
+ if (channel == null || !channel.isWritable()) {
+
source.fileMetricEventInc(StatConstants.EVENT_CHANNEL_NOT_WRITABLE);
+ if (logCounter.shouldPrint()) {
+ logger.warn("Prepare send msg but channel full, msgType={},
attr={}, channel={}",
+ msgObj.getMsgType(), msgObj.getAttr(), channel);
+ }
+ throw new Exception("Prepare send msg but channel full");
+ }
+ // check whether return response message
+ if (!msgObj.isNeedResp()) {
+ return;
+ }
+ // build return attribute string
+ strBuff.append(ConfigConstants.DATAPROXY_IP_KEY)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(source.getSrcHost());
+ if (msgObj.getErrCode() != DataProxyErrCode.SUCCESS) {
+
strBuff.append(AttributeConstants.SEPARATOR).append(AttributeConstants.MESSAGE_PROCESS_ERRCODE)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(msgObj.getErrCode().getErrCodeStr());
+ if (StringUtils.isNotEmpty(msgObj.getErrMsg())) {
+
strBuff.append(AttributeConstants.SEPARATOR).append(AttributeConstants.MESSAGE_PROCESS_ERRMSG)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(msgObj.getErrMsg());
+ }
+ if (StringUtils.isNotEmpty(msgObj.getAttr())) {
+
strBuff.append(AttributeConstants.SEPARATOR).append(msgObj.getAttr());
+ }
+ }
+ // build and send response message
+ ByteBuf retData;
+ MsgType msgType = MsgType.valueOf(msgObj.getMsgType());
+ if (MsgType.MSG_BIN_MULTI_BODY.equals(msgType)) {
+ retData = buildBinMsgRspPackage(strBuff.toString(),
msgObj.getUniq());
+ } else {
+ retData = buildTxtMsgRspPackage(msgType, strBuff.toString());
+ }
+ strBuff.delete(0, strBuff.length());
+ flushV0MsgPackage(source, channel, retData, msgObj.getAttr());
+ }
+
+ /**
+ * extract and process bin heart beat msg, message type is 8
+ */
+ private void procBinHeartbeatMsg(BaseSource source, Channel channel,
+ ByteBuf cb, int totalDataLen) throws Exception {
+ // Check if the message is complete and legal
+ if (totalDataLen < BIN_HB_FIXED_CONTENT_SIZE) {
+ source.fileMetricEventInc(StatConstants.EVENT_MALFORMED);
+ String errMsg = String.format("Malformed msg, totalDataLen(%d) <
min hb-msg length(%d)",
+ totalDataLen, BIN_HB_FIXED_CONTENT_SIZE);
+ if (logger.isDebugEnabled()) {
+ logger.debug(errMsg + ", channel {}", channel);
+ }
+ throw new Exception(errMsg);
+ }
+ // check validation
+ int msgHeadPos = cb.readerIndex() - 5;
+ int bodyLen = cb.getInt(msgHeadPos + BIN_HB_BODYLEN_OFFSET);
+ int attrLen = cb.getShort(msgHeadPos + BIN_HB_BODY_OFFSET + bodyLen);
+ int msgMagic = cb.getUnsignedShort(msgHeadPos
+ + BIN_HB_BODY_OFFSET + bodyLen + BIN_HB_ATTRLEN_SIZE +
attrLen);
+ if ((totalDataLen + BIN_HB_TOTALLEN_SIZE < (bodyLen + attrLen +
BIN_HB_FORMAT_SIZE))
+ || (msgMagic != BIN_MSG_MAGIC)) {
+ source.fileMetricEventInc(StatConstants.EVENT_MALFORMED);
+ String errMsg = String.format(
+ "Malformed msg, bodyLen(%d) + attrLen(%d) >
totalDataLen(%d) or msgMagic(%d) != %d",
+ bodyLen, attrLen, totalDataLen, msgMagic, BIN_MSG_MAGIC);
+ if (logger.isDebugEnabled()) {
+ logger.debug(errMsg + ", channel {}", channel);
+ }
+ throw new Exception(errMsg);
+ }
+ // read message content
+ byte version = cb.getByte(msgHeadPos + BIN_HB_VERSION_OFFSET);
+ byte[] attrData = null;
+ if (attrLen > 0) {
+ attrData = new byte[attrLen];
+ cb.getBytes(msgHeadPos + BIN_HB_BODY_OFFSET
+ + bodyLen + BIN_HB_ATTRLEN_SIZE, attrData, 0, attrLen);
+ }
+ // build and send response message
+ flushV0MsgPackage(source, channel, buildHBRspPackage(attrData,
version, 0),
+ MsgType.MSG_BIN_HEARTBEAT.name());
+ }
+
+ /**
+ * Build bin-msg response message ByteBuf
+ *
+ * @param attrs the return attribute
+ * @param uniqVal sequence Id
+ * @return ByteBuf
+ */
+ private ByteBuf buildBinMsgRspPackage(String attrs, long uniqVal) {
+ // calculate total length
+ // binTotalLen = mstType + uniq + attrsLen + attrs + magic
+ int binTotalLen = 1 + 4 + 2 + 2;
+ if (null != attrs) {
+ binTotalLen += attrs.length();
+ }
+ // allocate buffer and write fields
+ ByteBuf binBuffer = ByteBufAllocator.DEFAULT.buffer(4 + binTotalLen);
+ binBuffer.writeInt(binTotalLen);
+ binBuffer.writeByte(MsgType.MSG_BIN_MULTI_BODY.getValue());
+ byte[] uniq = new byte[4];
+ uniq[0] = (byte) ((uniqVal >> 24) & 0xFF);
+ uniq[1] = (byte) ((uniqVal >> 16) & 0xFF);
+ uniq[2] = (byte) ((uniqVal >> 8) & 0xFF);
+ uniq[3] = (byte) (uniqVal & 0xFF);
+ binBuffer.writeBytes(uniq);
+ if (null != attrs) {
+ binBuffer.writeShort(attrs.length());
+ binBuffer.writeBytes(attrs.getBytes(StandardCharsets.UTF_8));
+ } else {
+ binBuffer.writeShort(0x0);
+ }
+ binBuffer.writeShort(0xee01);
+ return binBuffer;
+ }
+
+ /**
+ * Build default-msg response message ByteBuf
+ *
+ * @param msgType the message type
+ * @param attrs the return attribute
+ * @return ByteBuf
+ */
+ private ByteBuf buildTxtMsgRspPackage(MsgType msgType, String attrs) {
+ int attrsLen = 0;
+ int bodyLen = 0;
+ if (attrs != null) {
+ attrsLen = attrs.length();
+ }
+ // backTotalLen = mstType + bodyLen + body + attrsLen + attrs
+ int backTotalLen = 1 + 4 + bodyLen + 4 + attrsLen;
+ ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(4 + backTotalLen);
+ buffer.writeInt(backTotalLen);
+ buffer.writeByte(msgType.getValue());
+ buffer.writeInt(bodyLen);
+ buffer.writeInt(attrsLen);
+ if (attrsLen > 0) {
+ buffer.writeBytes(attrs.getBytes(StandardCharsets.UTF_8));
+ }
+ return buffer;
+ }
+
+ /**
+ * Build heartbeat response message ByteBuf
+ *
+ * @param attrData the attribute data
+ * @param version the version
+ * @param loadValue the node load value
+ * @return ByteBuf
+ */
+ private ByteBuf buildHBRspPackage(byte[] attrData, byte version, int
loadValue) {
+ // calculate total length
+ // binTotalLen = mstType + dataTime + version + bodyLen + body +
attrsLen + attrs + magic
+ int binTotalLen = 1 + 4 + 1 + 4 + 2 + 2 + 2;
+ if (null != attrData) {
+ binTotalLen += attrData.length;
+ }
+ // check load value
+ if (loadValue == 0 || loadValue == (-1)) {
+ loadValue = 0xffff;
+ }
+ // allocate buffer and write fields
+ ByteBuf binBuffer = ByteBufAllocator.DEFAULT.buffer(4 + binTotalLen);
+ binBuffer.writeInt(binTotalLen);
+ binBuffer.writeByte(MsgType.MSG_BIN_HEARTBEAT.getValue());
+ binBuffer.writeInt((int) (System.currentTimeMillis() / 1000));
+ binBuffer.writeByte(version);
+ binBuffer.writeInt(2);
+ binBuffer.writeShort(loadValue);
+ if (null != attrData) {
+ binBuffer.writeShort(attrData.length);
+ binBuffer.writeBytes(attrData);
+ } else {
+ binBuffer.writeShort(0x0);
+ }
+ binBuffer.writeShort(0xee01);
+ return binBuffer;
+ }
+
+ /**
+ * Build hearbeat(1)-msg response message ByteBuf
+ *
+ * @return ByteBuf
+ */
+ private ByteBuf buildHeartBeatMsgRspPackage() {
+ ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(5);
+ // magic data
+ buffer.writeBytes(new byte[]{0, 0, 0, 1, 1});
+ return buffer;
+ }
+
+ private void flushV0MsgPackage(BaseSource source, Channel channel,
+ ByteBuf binBuffer, String orgAttr) throws Exception {
+ if (channel == null || !channel.isWritable()) {
+ // release allocated ByteBuf
+ binBuffer.release();
+
source.fileMetricEventInc(StatConstants.EVENT_CHANNEL_NOT_WRITABLE);
+ if (logCounter.shouldPrint()) {
+ logger.warn("Send msg but channel full, attr={}, channel={}",
orgAttr, channel);
+ }
+ throw new Exception("Send response but channel full");
+ }
+ channel.writeAndFlush(binBuffer);
+ }
+
+ /**
+ * addMetric
+ *
+ * @param result
+ * @param size
+ * @param event
+ */
+ private void addMetric(boolean result, long size, Event event) {
+ Map<String, String> dimensions = new HashMap<>();
+ dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
CommonConfigHolder.getInstance().getClusterName());
+ dimensions.put(DataProxyMetricItem.KEY_SOURCE_ID, source.getName());
+ dimensions.put(DataProxyMetricItem.KEY_SOURCE_DATA_ID,
source.getStrPort());
+ DataProxyMetricItem.fillInlongId(event, dimensions);
+ DataProxyMetricItem.fillAuditFormatTime(event, dimensions);
+ DataProxyMetricItem metricItem =
source.getMetricItemSet().findMetricItem(dimensions);
+ if (result) {
+ metricItem.readSuccessCount.incrementAndGet();
+ metricItem.readSuccessSize.addAndGet(size);
+ AuditUtils.add(AuditUtils.AUDIT_ID_DATAPROXY_READ_SUCCESS, event);
+ } else {
+ metricItem.readFailCount.incrementAndGet();
+ metricItem.readFailSize.addAndGet(size);
+ }
+ }
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SimpleTcpSource.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SimpleTcpSource.java
new file mode 100644
index 000000000..6be918b0e
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SimpleTcpSource.java
@@ -0,0 +1,153 @@
+/*
+ * 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.dataproxy.source2;
+
+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.ChannelOption;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import org.apache.flume.Context;
+import org.apache.flume.conf.Configurable;
+import org.apache.inlong.dataproxy.config.ConfigManager;
+import org.apache.inlong.dataproxy.config.holder.ConfigUpdateCallback;
+import org.apache.inlong.dataproxy.utils.AddressUtils;
+import org.apache.inlong.dataproxy.utils.EventLoopUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+
+/**
+ * Simple tcp source
+ *
+ */
+public class SimpleTcpSource extends BaseSource implements Configurable,
ConfigUpdateCallback {
+
+ private static final Logger logger =
LoggerFactory.getLogger(SimpleTcpSource.class);
+
+ private ServerBootstrap bootstrap;
+ private boolean tcpNoDelay;
+ private boolean tcpKeepAlive;
+ private int highWaterMark;
+ private boolean enableBusyWait;
+
+ public SimpleTcpSource() {
+ super();
+ ConfigManager.getInstance().regIPVisitConfigChgCallback(this);
+ }
+
+ @Override
+ public void configure(Context context) {
+ logger.info("Source {} context is {}", getName(), context);
+ super.configure(context);
+ // get tcp no-delay parameter
+ this.tcpNoDelay =
context.getBoolean(SourceConstants.SRCCXT_TCP_NO_DELAY,
+ SourceConstants.VAL_DEF_TCP_NO_DELAY);
+ // get tcp keep-alive parameter
+ this.tcpKeepAlive =
context.getBoolean(SourceConstants.SRCCXT_TCP_KEEP_ALIVE,
+ SourceConstants.VAL_DEF_TCP_KEEP_ALIVE);
+ // get tcp enable busy-wait
+ this.enableBusyWait =
context.getBoolean(SourceConstants.SRCCXT_TCP_ENABLE_BUSY_WAIT,
+ SourceConstants.VAL_DEF_TCP_ENABLE_BUSY_WAIT);
+ // get tcp high watermark
+ this.highWaterMark = getIntValue(context,
SourceConstants.SRCCXT_TCP_HIGH_WATER_MARK,
+ SourceConstants.VAL_DEF_TCP_HIGH_WATER_MARK);
+ Preconditions.checkArgument((this.highWaterMark >=
SourceConstants.VAL_MIN_TCP_HIGH_WATER_MARK),
+ SourceConstants.VAL_DEF_TCP_HIGH_WATER_MARK + " must be >= "
+ + SourceConstants.VAL_MIN_TCP_HIGH_WATER_MARK);
+ }
+
+ @Override
+ public synchronized void startSource() {
+ logger.info("start " + this.getName());
+ // build accept group
+ this.acceptorGroup = EventLoopUtil.newEventLoopGroup(maxAcceptThreads,
enableBusyWait,
+ new DefaultThreadFactory(this.getName() + "-boss-group"));
+ // build worker group
+ this.workerGroup = EventLoopUtil.newEventLoopGroup(maxWorkerThreads,
enableBusyWait,
+ new DefaultThreadFactory(this.getName() + "-worker-group"));
+ // init boostrap
+ bootstrap = new ServerBootstrap();
+ bootstrap.childOption(ChannelOption.ALLOCATOR,
ByteBufAllocator.DEFAULT);
+ bootstrap.childOption(ChannelOption.TCP_NODELAY, tcpNoDelay);
+ bootstrap.childOption(ChannelOption.SO_KEEPALIVE, tcpKeepAlive);
+ bootstrap.childOption(ChannelOption.SO_RCVBUF, maxRcvBufferSize);
+ bootstrap.childOption(ChannelOption.SO_SNDBUF, maxSendBufferSize);
+ bootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK,
highWaterMark);
+
bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(workerGroup));
+ EventLoopUtil.enableTriggeredMode(bootstrap);
+ bootstrap.group(acceptorGroup, workerGroup);
+ bootstrap.childHandler(this.getChannelInitializerFactory());
+ try {
+ if (srcHost == null) {
+ channelFuture = bootstrap.bind(new
InetSocketAddress(srcPort)).sync();
+ } else {
+ channelFuture = bootstrap.bind(new InetSocketAddress(srcHost,
srcPort)).sync();
+ }
+ } catch (Exception e) {
+ logger.error("Source {} bind ({}:{}) error, program will exit! e =
{}",
+ this.getName(), srcHost, srcPort, e);
+ System.exit(-1);
+ }
+ ConfigManager.getInstance().addSourceReportInfo(
+ srcHost, String.valueOf(srcPort),
getProtocolName().toUpperCase());
+ logger.info("Source {} started at ({}:{})!", this.getName(), srcHost,
srcPort);
+ }
+
+ @Override
+ public synchronized void stop() {
+ super.stop();
+ }
+
+ @Override
+ public String getProtocolName() {
+ return SourceConstants.SRC_PROTOCOL_TYPE_TCP;
+ }
+
+ @Override
+ public void update() {
+ // check current all links
+ if (ConfigManager.getInstance().needChkIllegalIP()) {
+ int cnt = 0;
+ Channel channel;
+ String strRemoteIP;
+ long startTime = System.currentTimeMillis();
+ Iterator<Channel> iterator = allChannels.iterator();
+ while (iterator.hasNext()) {
+ channel = iterator.next();
+ strRemoteIP = AddressUtils.getChannelRemoteIP(channel);
+ if (strRemoteIP == null) {
+ continue;
+ }
+ if (ConfigManager.getInstance().isIllegalIP(strRemoteIP)) {
+ channel.disconnect();
+ channel.close();
+ allChannels.remove(channel);
+ cnt++;
+ logger.error(strRemoteIP + " is Illegal IP, so disconnect
it !");
+ }
+ }
+ logger.info("Source {} channel check, disconnects {} Illegal
channels, waist {} ms",
+ getName(), cnt, (System.currentTimeMillis() - startTime));
+ }
+ }
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SimpleUdpSource.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SimpleUdpSource.java
new file mode 100644
index 000000000..dbe988f68
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SimpleUdpSource.java
@@ -0,0 +1,82 @@
+/*
+ * 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.dataproxy.source2;
+
+import org.apache.flume.Context;
+import org.apache.flume.conf.Configurable;
+import org.apache.inlong.dataproxy.config.ConfigManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.net.InetSocketAddress;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+
+public class SimpleUdpSource extends BaseSource implements Configurable {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(SimpleUdpSource.class);
+
+ private Bootstrap bootstrap;
+
+ public SimpleUdpSource() {
+ super();
+ }
+
+ @Override
+ public void configure(Context context) {
+ logger.info("Source {} context is {}", getName(), context);
+ super.configure(context);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ @Override
+ public void startSource() {
+ // setup Netty server
+ logger.info("start " + this.getName());
+ bootstrap = new Bootstrap();
+ bootstrap.channel(NioDatagramChannel.class);
+ bootstrap.option(ChannelOption.SO_RCVBUF, maxRcvBufferSize);
+ bootstrap.option(ChannelOption.SO_SNDBUF, maxSendBufferSize);
+ bootstrap.handler(this.getChannelInitializerFactory());
+ try {
+ if (srcHost == null) {
+ channelFuture = bootstrap.bind(new
InetSocketAddress(srcPort)).sync();
+ } else {
+ channelFuture = bootstrap.bind(new InetSocketAddress(srcHost,
srcPort)).sync();
+ }
+ } catch (Exception e) {
+ logger.error("Source {} bind ({}:{}) error, program will exit! e =
{}",
+ this.getName(), srcHost, srcPort, e);
+ System.exit(-1);
+ }
+ ConfigManager.getInstance().addSourceReportInfo(
+ srcHost, String.valueOf(srcPort),
getProtocolName().toUpperCase());
+ logger.info("Source {} started at ({}:{})!", this.getName(), srcHost,
srcPort);
+ }
+
+ @Override
+ public void stop() {
+ super.stop();
+ }
+
+ @Override
+ public String getProtocolName() {
+ return SourceConstants.SRC_PROTOCOL_TYPE_UDP;
+ }
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SourceConstants.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SourceConstants.java
new file mode 100644
index 000000000..edb111dd8
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SourceConstants.java
@@ -0,0 +1,202 @@
+/*
+ * 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.dataproxy.source2;
+
+public class SourceConstants {
+
+ // source host
+ public static final String SRCCXT_CONFIG_HOST = "host";
+ // system env source host
+ public static final String SYSENV_HOST_IP = "inlongHostIp";
+ // default source host
+ public static final String VAL_DEF_HOST_VALUE = "0.0.0.0";
+ // source port
+ public static final String SRCCXT_CONFIG_PORT = "port";
+ // system env source port
+ public static final String SYSENV_HOST_PORT = "inlongHostPort";
+ // message factory name
+ public static final String SRCCXT_MSG_FACTORY_NAME = "msg-factory-name";
+ // message handler name
+ public static final String SRCCXT_MESSAGE_HANDLER_NAME =
"message-handler-name";
+ // default topic name
+ public static final String SRCCXT_DEF_TOPIC = "topic";
+ // default attributes
+ public static final String SRCCXT_DEF_ATTR = "attr";
+ // max message length
+ public static final String SRCCXT_MAX_MSG_LENGTH = "max-msg-length";
+ // allowed max message length
+ public static final int VAL_MAX_MAX_MSG_LENGTH = 20 * 1024 * 1024;
+ public static final int VAL_MIN_MAX_MSG_LENGTH = 5;
+ public static final int VAL_DEF_MAX_MSG_LENGTH = 1024 * 64;
+ // whether compress message
+ public static final String SRCCXT_MSG_COMPRESSED = "msg-compressed";
+ public static final boolean VAL_DEF_MSG_COMPRESSED = true;
+ // whether filter empty message
+ public static final String SRCCXT_FILTER_EMPTY_MSG = "filter-empty-msg";
+ public static final boolean VAL_DEF_FILTER_EMPTY_MSG = false;
+ // whether custom channel processor
+ public static final String SRCCXT_CUSTOM_CHANNEL_PROCESSOR = "custom-cp";
+ public static final boolean VAL_DEF_CUSTOM_CH_PROCESSOR = false;
+ // max net accept process threads
+ public static final String SRCCXT_MAX_ACCEPT_THREADS =
"max-accept-threads";
+ public static final int VAL_DEF_NET_ACCEPT_THREADS = 1;
+ public static final int VAL_MIN_ACCEPT_THREADS = 1;
+ public static final int VAL_MAX_ACCEPT_THREADS = 10;
+ // max net worker process threads
+ public static final String SRCCXT_MAX_WORKER_THREADS = "max-threads";
+ public static final int VAL_DEF_WORKER_THREADS =
Runtime.getRuntime().availableProcessors();
+ public static final int VAL_MIN_WORKER_THREADS = 1;
+ public static final int VAL_MAX_WORKER_THREADS =
Runtime.getRuntime().availableProcessors() * 2;
+ // file metric statistic interval(second)
+ public static final String SRCCXT_STAT_INTERVAL_SEC = "stat-interval-sec";
+ public static final int VAL_DEF_STAT_INVL_SEC = 60;
+ public static final int VAL_MIN_STAT_INVL_SEC = 0;
+ // max file statistic key count
+ public static final String SRCCXT_MAX_MONITOR_STAT_CNT = "max-monitor-cnt";
+ public static final int VAL_DEF_MON_STAT_CNT = 1000000;
+ public static final int VAL_MIN_MON_STAT_CNT = 0;
+ // max file statistic key count
+ public static final String SRCCXT_FILE_METRIC_ON = "file-metric-on";
+ public static final boolean VAL_DEF_FILE_METRIC_ON = true;
+ // max connection count
+ public static final String SRCCXT_MAX_CONNECTION_CNT = "connections";
+ public static final int VAL_DEF_MAX_CONNECTION_CNT = 5000;
+ public static final int VAL_MIN_CONNECTION_CNT = 0;
+ // max receive buffer size
+ public static final String SRCCXT_RECEIVE_BUFFER_SIZE =
"receiveBufferSize";
+ public static final int VAL_DEF_RECEIVE_BUFFER_SIZE = 64 * 1024;
+ public static final int VAL_MIN_RECEIVE_BUFFER_SIZE = 0;
+ public static final int VAL_MAX_RECEIVE_BUFFER_SIZE = 100 * 1024 * 1024;
+ // max send buffer size
+ public static final String SRCCXT_SEND_BUFFER_SIZE = "sendBufferSize";
+ public static final int VAL_DEF_SEND_BUFFER_SIZE = 64 * 1024;
+ public static final int VAL_MIN_SEND_BUFFER_SIZE = 0;
+ public static final int VAL_MAX_SEND_BUFFER_SIZE = 100 * 1024 * 1024;
+ // tcp parameter no delay
+ public static final String SRCCXT_TCP_NO_DELAY = "tcpNoDelay";
+ public static final boolean VAL_DEF_TCP_NO_DELAY = true;
+ // tcp parameter keep alive
+ public static final String SRCCXT_TCP_KEEP_ALIVE = "keepAlive";
+ public static final boolean VAL_DEF_TCP_KEEP_ALIVE = true;
+ // tcp parameter high water mark
+ public static final String SRCCXT_TCP_HIGH_WATER_MARK = "highWaterMark";
+ public static final int VAL_DEF_TCP_HIGH_WATER_MARK = 64 * 1024;
+ public static final int VAL_MIN_TCP_HIGH_WATER_MARK = 0;
+ // tcp parameter enable busy wait
+ public static final String SRCCXT_TCP_ENABLE_BUSY_WAIT = "enableBusyWait";
+ public static final boolean VAL_DEF_TCP_ENABLE_BUSY_WAIT = false;
+ // tcp parameters max read idle time
+ public static final String SRCCXT_MAX_READ_IDLE_TIME_MS =
"maxReadIdleTime";
+ public static final long VAL_DEF_READ_IDLE_TIME_MS = 70 * 60 * 1000;
+ public static final long VAL_MIN_READ_IDLE_TIME_MS = 60 * 1000;
+ // source protocol type
+ public static final String SRC_PROTOCOL_TYPE_TCP = "tcp";
+ public static final String SRC_PROTOCOL_TYPE_UDP = "udp";
+ public static final String SRC_PROTOCOL_TYPE_HTTP = "http";
+
+ public static final String SERVICE_PROCESSOR_NAME = "service-decoder-name";
+ public static final String ENABLE_EXCEPTION_RETURN =
"enableExceptionReturn";
+
+ public static final String TRAFFIC_CLASS = "trafficClass";
+
+ public static final String HEART_INTERVAL_SEC = "heart-interval-sec";
+
+ public static final String PACKAGE_TIMEOUT_SEC = "package-timeout-sec";
+
+ public static final String HEART_SERVERS = "heart-servers";
+
+ public static final String TOPIC_KEY = "topic";
+ public static final String REMOTE_IP_KEY = "srcIp";
+ public static final String DATAPROXY_IP_KEY = "dpIp";
+ public static final String MSG_ENCODE_VER = "msgEnType";
+ public static final String REMOTE_IDC_KEY = "idc";
+ public static final String MSG_COUNTER_KEY = "msgcnt";
+ public static final String PKG_COUNTER_KEY = "pkgcnt";
+ public static final String PKG_TIME_KEY = "msg.pkg.time";
+ public static final String TRANSFER_KEY = "transfer";
+ public static final String DEST_IP_KEY = "dstIp";
+ public static final String INTERFACE_KEY = "interface";
+ public static final String SINK_MIN_METRIC_KEY = "sink-min-metric-topic";
+ public static final String SINK_HOUR_METRIC_KEY = "sink-hour-metric-topic";
+ public static final String SINK_TEN_METRIC_KEY = "sink-ten-metric-topic";
+ public static final String SINK_QUA_METRIC_KEY = "sink-qua-metric-topic";
+ public static final String L5_MIN_METRIC_KEY = "l5-min-metric-topic";
+ public static final String L5_MIN_FAIL_METRIC_KEY =
"l5-min-fail-metric-key";
+ public static final String L5_HOUR_METRIC_KEY = "l5-hour-metric-topic";
+ public static final String L5_ID_KEY = "l5id";
+ public static final String SET_KEY = "set";
+ public static final String CLUSTER_ID_KEY = "clusterId";
+
+ public static final String DECODER_BODY = "body";
+ public static final String DECODER_TOPICKEY = "topic_key";
+ public static final String DECODER_ATTRS = "attrs";
+ public static final String MSG_TYPE = "msg_type";
+ public static final String COMPRESS_TYPE = "compress_type";
+ public static final String EXTRA_ATTR = "extra_attr";
+ public static final String COMMON_ATTR_MAP = "common_attr_map";
+ public static final String MSG_LIST = "msg_list";
+ public static final String VERSION_TYPE = "version";
+ public static final String FILE_CHECK_DATA = "file-check-data";
+ public static final String MINUTE_CHECK_DATA = "minute-check-data";
+ public static final String SLA_METRIC_DATA = "sla-metric-data";
+ public static final String SLA_METRIC_GROUPID = "manager_sla_metric";
+
+ public static final String FILE_BODY = "file-body";
+ public static final int MSG_MAX_LENGTH_BYTES = 20 * 1024 * 1024;
+
+ public static final String SEQUENCE_ID = "sequencial_id";
+
+ public static final String TOTAL_LEN = "totalLen";
+
+ public static final String LINK_MAX_ALLOWED_DELAYED_MSG_COUNT =
"link_max_allowed_delayed_msg_count";
+ public static final String SESSION_WARN_DELAYED_MSG_COUNT =
"session_warn_delayed_msg_count";
+ public static final String SESSION_MAX_ALLOWED_DELAYED_MSG_COUNT =
"session_max_allowed_delayed_msg_count";
+ public static final String NETTY_WRITE_BUFFER_HIGH_WATER_MARK =
"netty_write_buffer_high_water_mark";
+ public static final String RECOVER_THREAD_COUNT = "recover_thread_count";
+
+ public static final String MANAGER_PATH = "/inlong/manager/openapi";
+ public static final String MANAGER_GET_CONFIG_PATH =
"/dataproxy/getConfig";
+ public static final String MANAGER_GET_ALL_CONFIG_PATH =
"/dataproxy/getAllConfig";
+ public static final String MANAGER_HEARTBEAT_REPORT = "/heartbeat/report";
+
+ public static final String MANAGER_AUTH_SECRET_ID =
"manager.auth.secretId";
+ public static final String MANAGER_AUTH_SECRET_KEY =
"manager.auth.secretKey";
+ // Pulsar config
+ public static final String KEY_TENANT = "tenant";
+ public static final String KEY_NAMESPACE = "namespace";
+
+ public static final String KEY_SERVICE_URL = "serviceUrl";
+ public static final String KEY_AUTHENTICATION = "authentication";
+ public static final String KEY_STATS_INTERVAL_SECONDS =
"statsIntervalSeconds";
+
+ public static final String KEY_ENABLEBATCHING = "enableBatching";
+ public static final String KEY_BATCHINGMAXBYTES = "batchingMaxBytes";
+ public static final String KEY_BATCHINGMAXMESSAGES = "batchingMaxMessages";
+ public static final String KEY_BATCHINGMAXPUBLISHDELAY =
"batchingMaxPublishDelay";
+ public static final String KEY_MAXPENDINGMESSAGES = "maxPendingMessages";
+ public static final String KEY_MAXPENDINGMESSAGESACROSSPARTITIONS =
"maxPendingMessagesAcrossPartitions";
+ public static final String KEY_SENDTIMEOUT = "sendTimeout";
+ public static final String KEY_COMPRESSIONTYPE = "compressionType";
+ public static final String KEY_BLOCKIFQUEUEFULL = "blockIfQueueFull";
+ public static final String
KEY_ROUNDROBINROUTERBATCHINGPARTITIONSWITCHFREQUENCY = "roundRobinRouter"
+ + "BatchingPartitionSwitchFrequency";
+
+ public static final String KEY_IOTHREADS = "ioThreads";
+ public static final String KEY_MEMORYLIMIT = "memoryLimit";
+ public static final String KEY_CONNECTIONSPERBROKER =
"connectionsPerBroker";
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/AbsV0MsgCodec.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/AbsV0MsgCodec.java
new file mode 100644
index 000000000..200228ca8
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/AbsV0MsgCodec.java
@@ -0,0 +1,223 @@
+/*
+ * 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.dataproxy.source2.v0msg;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flume.Event;
+import org.apache.inlong.common.enums.DataProxyErrCode;
+import org.apache.inlong.common.msg.AttributeConstants;
+import org.apache.inlong.dataproxy.consts.ConfigConstants;
+import org.apache.inlong.dataproxy.consts.StatConstants;
+import org.apache.inlong.dataproxy.source2.BaseSource;
+import org.apache.inlong.dataproxy.utils.DateTimeUtils;
+import org.apache.inlong.dataproxy.utils.InLongMsgVer;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+
+public abstract class AbsV0MsgCodec {
+
+ // string splitter
+ protected static final Splitter.MapSplitter mapSplitter = Splitter
+ .on(AttributeConstants.SEPARATOR).trimResults()
+ .withKeyValueSeparator(AttributeConstants.KEY_VALUE_SEPARATOR);
+ // map joiner
+ protected static final Joiner.MapJoiner mapJoiner =
Joiner.on(AttributeConstants.SEPARATOR)
+ .withKeyValueSeparator(AttributeConstants.KEY_VALUE_SEPARATOR);
+
+ protected DataProxyErrCode errCode = DataProxyErrCode.UNKNOWN_ERROR;
+ protected String errMsg = "";
+ protected String strRemoteIP;
+ protected long msgRcvTime;
+ protected int totalDataLen;
+ protected byte msgType;
+ protected int msgCount;
+ protected String origAttr = "";
+ protected byte[] bodyData;
+ protected long dataTimeMs;
+ protected String groupId;
+ protected String streamId = "";
+ protected String topicName;
+ protected String msgSeqId = "";
+ protected long uniq = -1L;
+ protected String msgProcType = "b2b";
+ protected boolean needResp = true;
+ protected final Map<String, String> attrMap = new HashMap<>();
+
+ public AbsV0MsgCodec(int totalDataLen, int msgTypeValue,
+ long msgRcvTime, String strRemoteIP) {
+ this.totalDataLen = totalDataLen;
+ this.msgType = (byte) (msgTypeValue & 0xFF);
+ this.msgRcvTime = msgRcvTime;
+ this.strRemoteIP = strRemoteIP;
+ }
+
+ public abstract boolean descMsg(BaseSource source, ByteBuf cb) throws
Exception;
+
+ public abstract boolean validAndFillFields(BaseSource source,
StringBuilder strBuff);
+
+ public abstract Event encEventPackage(BaseSource source, Channel channel);
+
+ public DataProxyErrCode getErrCode() {
+ return this.errCode;
+ }
+
+ public String getErrMsg() {
+ return this.errMsg;
+ }
+
+ public boolean isNeedResp() {
+ return this.needResp;
+ }
+
+ public byte getMsgType() {
+ return this.msgType;
+ }
+
+ public String getAttr() {
+ return this.origAttr;
+ }
+
+ public Map<String, String> getAttrMap() {
+ return this.attrMap;
+ }
+
+ public long getUniq() {
+ return this.uniq;
+ }
+
+ public long getDataTimeMs() {
+ return this.dataTimeMs;
+ }
+
+ public String getGroupId() {
+ return this.groupId;
+ }
+
+ public String getStreamId() {
+ return this.streamId;
+ }
+
+ public String getTopicName() {
+ return this.topicName;
+ }
+
+ public String getMsgProcType() {
+ return this.msgProcType;
+ }
+
+ public int getBodyLength() {
+ return this.bodyData == null ? 0 : this.bodyData.length;
+ }
+
+ public int getMsgCount() {
+ return this.msgCount;
+ }
+
+ public String getStrRemoteIP() {
+ return strRemoteIP;
+ }
+
+ public long getMsgRcvTime() {
+ return msgRcvTime;
+ }
+
+ public void setFailureInfo(DataProxyErrCode errCode) {
+ setFailureInfo(errCode, "");
+ }
+
+ public void setFailureInfo(DataProxyErrCode errCode, String errMsg) {
+ this.errCode = errCode;
+ this.errMsg = errMsg;
+ }
+
+ protected boolean decAttrInfo(BaseSource source, ByteBuf cb,
+ int attrLen, int attrPos) throws Exception {
+ // get attr bytes
+ if (attrLen > 0) {
+ byte[] attrData = new byte[attrLen];
+ cb.getBytes(attrPos, attrData, 0, attrLen);
+ try {
+ this.origAttr = new String(attrData, StandardCharsets.UTF_8);
+ } catch (Throwable err) {
+ //
+ }
+ }
+ // parse attribute field
+ if (StringUtils.isNotBlank(this.origAttr)) {
+ try {
+ this.attrMap.putAll(mapSplitter.split(this.origAttr));
+ } catch (Exception e) {
+ source.fileMetricEventInc(StatConstants.EVENT_INVALIDATTR);
+ this.errCode = DataProxyErrCode.SPLIT_ATTR_ERROR;
+ this.errMsg = String.format("Parse attr (%s) failure",
this.origAttr);
+ return false;
+ }
+ }
+ // get whether return request
+ if
("false".equalsIgnoreCase(attrMap.get(AttributeConstants.MESSAGE_IS_ACK))) {
+ this.needResp = false;
+ }
+ return true;
+ }
+
+ protected Map<String, String> buildEventHeaders(long pkgTime) {
+ // build headers
+ Map<String, String> headers = new HashMap<>();
+ headers.put(AttributeConstants.GROUP_ID, groupId);
+ headers.put(AttributeConstants.STREAM_ID, streamId);
+ headers.put(ConfigConstants.TOPIC_KEY, topicName);
+ headers.put(AttributeConstants.DATA_TIME, String.valueOf(dataTimeMs));
+ headers.put(ConfigConstants.REMOTE_IP_KEY, strRemoteIP);
+ headers.put(ConfigConstants.MSG_COUNTER_KEY, String.valueOf(msgCount));
+ headers.put(ConfigConstants.MSG_ENCODE_VER,
InLongMsgVer.INLONG_V0.getName());
+ headers.put(AttributeConstants.RCV_TIME, String.valueOf(msgRcvTime));
+ // add extra key-value information
+ String pkgTimeStr = attrMap.get(ConfigConstants.PKG_TIME_KEY);
+ if (StringUtils.isBlank(pkgTimeStr)) {
+ pkgTimeStr = DateTimeUtils.ms2yyyyMMddHHmm(pkgTime);
+ }
+ headers.put(ConfigConstants.PKG_TIME_KEY, pkgTimeStr);
+ if (!needResp) {
+ headers.put(AttributeConstants.MESSAGE_IS_ACK, "false");
+ }
+ String syncSend = attrMap.get(AttributeConstants.MESSAGE_SYNC_SEND);
+ if (StringUtils.isNotEmpty(syncSend)) {
+ headers.put(AttributeConstants.MESSAGE_SYNC_SEND, syncSend);
+ }
+ String proxySend = attrMap.get(AttributeConstants.MESSAGE_PROXY_SEND);
+ if (StringUtils.isNotEmpty(proxySend)) {
+ headers.put(AttributeConstants.MESSAGE_PROXY_SEND, proxySend);
+ }
+ String partitionKey =
attrMap.get(AttributeConstants.MESSAGE_PARTITION_KEY);
+ if (StringUtils.isNotEmpty(partitionKey)) {
+ headers.put(AttributeConstants.MESSAGE_PARTITION_KEY,
partitionKey);
+ }
+ if (StringUtils.isNotEmpty(this.msgSeqId)) {
+ headers.put(ConfigConstants.SEQUENCE_ID, this.msgSeqId);
+ }
+ return headers;
+ }
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/CodecBinMsg.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/CodecBinMsg.java
new file mode 100644
index 000000000..a93a4fbae
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/CodecBinMsg.java
@@ -0,0 +1,363 @@
+/*
+ * 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.dataproxy.source2.v0msg;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.flume.Event;
+import org.apache.flume.event.EventBuilder;
+import org.apache.inlong.common.enums.DataProxyErrCode;
+import org.apache.inlong.common.msg.AttributeConstants;
+import org.apache.inlong.common.msg.InLongMsg;
+import org.apache.inlong.common.msg.MsgType;
+import org.apache.inlong.dataproxy.base.SinkRspEvent;
+import org.apache.inlong.dataproxy.config.CommonConfigHolder;
+import org.apache.inlong.dataproxy.config.ConfigManager;
+import org.apache.inlong.dataproxy.consts.StatConstants;
+import org.apache.inlong.dataproxy.source2.BaseSource;
+import org.apache.inlong.dataproxy.utils.MessageUtils;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_ATTRLEN_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_BODYLEN_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_BODY_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_CNT_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_DT_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_EXTEND_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_FORMAT_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_GROUPIDNUM_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_MAGIC;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_MAGIC_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_MSGTYPE_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_STREAMIDNUM_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_TOTALLEN_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_TOTALLEN_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.BIN_MSG_UNIQ_OFFSET;
+
+public class CodecBinMsg extends AbsV0MsgCodec {
+
+ private int groupIdNum;
+ private int streamIdNum;
+ private int extendField;
+ private long dataTimeSec;
+ private boolean num2name = false;
+ private boolean transNum2Name = false;
+ private boolean isOrderOrProxy = false;
+ private boolean indexMsg = false;
+ private boolean fileCheckMsg = false;
+ private boolean needTraceMsg = false;
+
+ public CodecBinMsg(int totalDataLen, int msgTypeValue,
+ long msgRcvTime, String strRemoteIP) {
+ super(totalDataLen, msgTypeValue, msgRcvTime, strRemoteIP);
+ }
+
+ public boolean descMsg(BaseSource source, ByteBuf cb) throws Exception {
+ int msgHeadPos = cb.readerIndex() - 5;
+ // read fixed field value
+ this.groupIdNum = cb.getUnsignedShort(BIN_MSG_GROUPIDNUM_OFFSET);
+ this.streamIdNum = cb.getUnsignedShort(BIN_MSG_STREAMIDNUM_OFFSET);
+ this.extendField = cb.getUnsignedShort(BIN_MSG_EXTEND_OFFSET);
+ this.dataTimeSec = cb.getUnsignedInt(BIN_MSG_DT_OFFSET);
+ this.dataTimeMs = this.dataTimeSec * 1000;
+ this.msgCount = cb.getUnsignedShort(BIN_MSG_CNT_OFFSET);
+ this.msgCount = (this.msgCount != 0) ? this.msgCount : 1;
+ this.uniq = cb.getUnsignedInt(BIN_MSG_UNIQ_OFFSET);
+ // get body and attribute field length
+ int bodyLen = cb.getInt(msgHeadPos + BIN_MSG_BODYLEN_OFFSET);
+ int attrLen = cb.getShort(msgHeadPos + BIN_MSG_BODY_OFFSET + bodyLen);
+ int msgMagic = cb.getUnsignedShort(msgHeadPos + BIN_MSG_BODY_OFFSET
+ + bodyLen + BIN_MSG_ATTRLEN_SIZE + attrLen);
+ if (bodyLen <= 0) {
+ if (bodyLen == 0) {
+ source.fileMetricEventInc(StatConstants.EVENT_NOBODY);
+ this.errCode = DataProxyErrCode.BODY_LENGTH_ZERO;
+ } else {
+ source.fileMetricEventInc(StatConstants.EVENT_NEGBODY);
+ this.errCode = DataProxyErrCode.BODY_LENGTH_LESS_ZERO;
+ }
+ return false;
+ }
+ // get attribute length
+ if (attrLen < 0) {
+ source.fileMetricEventInc(StatConstants.EVENT_NEGATTR);
+ this.errCode = DataProxyErrCode.ATTR_LENGTH_LESS_ZERO;
+ return false;
+ }
+ // get msg magic
+ if ((msgMagic != BIN_MSG_MAGIC)
+ || (totalDataLen + BIN_MSG_TOTALLEN_SIZE < (bodyLen + attrLen
+ BIN_MSG_FORMAT_SIZE))) {
+ source.fileMetricEventInc(StatConstants.EVENT_MALFORMED);
+ this.errCode = DataProxyErrCode.FIELD_VALUE_NOT_EQUAL;
+ this.errMsg = String.format(
+ "fixedLen(%d) + bodyLen(%d) + attrLen(%d) >
totalDataLen(%d) + 4 or msgMagic(%d) != %d",
+ BIN_MSG_FORMAT_SIZE, bodyLen, attrLen, totalDataLen,
msgMagic, BIN_MSG_MAGIC);
+ return false;
+ }
+ // extract attr bytes
+ if (!decAttrInfo(source, cb, attrLen,
+ msgHeadPos + BIN_MSG_BODY_OFFSET + bodyLen +
BIN_MSG_ATTRLEN_SIZE)) {
+ return false;
+ }
+ this.bodyData = new byte[bodyLen];
+ cb.getBytes(msgHeadPos + BIN_MSG_BODY_OFFSET, this.bodyData, 0,
bodyLen);
+ // process extend field value
+ if (((this.extendField & 0x8) == 0x8) || ((this.extendField & 0x10) ==
0x10)) {
+ this.indexMsg = true;
+ this.fileCheckMsg = (this.extendField & 0x8) == 0x8;
+ }
+ if (((extendField & 0x2) >> 1) == 0x1) {
+ this.needTraceMsg = true;
+ }
+ if (((extendField & 0x4) >> 2) == 0x0) {
+ this.num2name = true;
+ }
+ // parse required fields
+ Pair<Boolean, String> evenProcType =
+
MessageUtils.getEventProcType(attrMap.get(AttributeConstants.MESSAGE_SYNC_SEND),
+ attrMap.get(AttributeConstants.MESSAGE_PROXY_SEND));
+ this.isOrderOrProxy = evenProcType.getLeft();
+ this.msgProcType = evenProcType.getRight();
+ return true;
+ }
+
+ public boolean validAndFillFields(BaseSource source, StringBuilder
strBuff) {
+ // reject unsupported index messages
+ if (indexMsg) {
+ source.fileMetricEventInc(StatConstants.EVENT_UNSUPMSG);
+ this.errCode = DataProxyErrCode.UNSUPPORTED_EXTEND_FIELD_VALUE;
+ return false;
+ }
+ // valid and fill topicName
+ if (!validAndFillTopic(source)) {
+ return false;
+ }
+ // build message seqId
+ this.msgSeqId = strBuff.append(this.topicName)
+ .append(AttributeConstants.SEPARATOR).append(this.streamId)
+ .append(AttributeConstants.SEPARATOR).append(strRemoteIP)
+
.append("#").append(dataTimeMs).append("#").append(uniq).toString();
+ strBuff.delete(0, strBuff.length());
+ // check required rtms attrs
+ if (StringUtils.isBlank(attrMap.get(AttributeConstants.MSG_RPT_TIME)))
{
+ strBuff.append(AttributeConstants.MSG_RPT_TIME)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(msgRcvTime);
+ attrMap.put(AttributeConstants.MSG_RPT_TIME,
String.valueOf(msgRcvTime));
+ }
+ // get trace requirement
+ if (this.needTraceMsg) {
+ if (strBuff.length() > 0) {
+ strBuff.append(AttributeConstants.SEPARATOR);
+ }
+ strBuff.append(AttributeConstants.DATAPROXY_NODE_IP)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(source.getStrPort())
+ .append(AttributeConstants.SEPARATOR)
+ .append(AttributeConstants.DATAPROXY_RCVTIME)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(msgRcvTime);
+ attrMap.put(AttributeConstants.DATAPROXY_NODE_IP,
source.getSrcHost());
+ attrMap.put(AttributeConstants.DATAPROXY_RCVTIME,
String.valueOf(msgRcvTime));
+ }
+ // trans groupId and StreamId Num 2 Name
+ if (this.transNum2Name) {
+ if (strBuff.length() > 0) {
+ strBuff.append(AttributeConstants.SEPARATOR);
+ }
+ strBuff.append(AttributeConstants.GROUP_ID)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(groupId)
+ .append(AttributeConstants.SEPARATOR)
+ .append(AttributeConstants.STREAM_ID)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(streamId);
+ for (Map.Entry<String, String> entry : attrMap.entrySet()) {
+ if
(AttributeConstants.GROUP_ID.equalsIgnoreCase(entry.getKey())
+ ||
AttributeConstants.STREAM_ID.equalsIgnoreCase(entry.getKey())) {
+ continue;
+ }
+ strBuff.append(AttributeConstants.SEPARATOR)
+ .append(entry.getKey())
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(entry.getValue());
+ }
+ this.groupIdNum = 0;
+ this.streamIdNum = 0;
+ this.num2name = false;
+ this.extendField = this.extendField | 0x4;
+ attrMap.put(AttributeConstants.GROUP_ID, groupId);
+ attrMap.put(AttributeConstants.STREAM_ID, streamId);
+ }
+ if (strBuff.length() > 0) {
+ if (StringUtils.isNotBlank(origAttr)) {
+ strBuff.append(AttributeConstants.SEPARATOR).append(origAttr);
+ }
+ totalDataLen += strBuff.length() - origAttr.length();
+ origAttr = strBuff.toString();
+ strBuff.delete(0, strBuff.length());
+ }
+ return true;
+ }
+
+ public Event encEventPackage(BaseSource source, Channel channel) {
+ // fill bin msg package
+ int totalPkgLength = totalDataLen + BIN_MSG_TOTALLEN_SIZE;
+ ByteBuffer dataBuf = ByteBuffer.allocate(totalPkgLength);
+ dataBuf.putInt(BIN_MSG_TOTALLEN_OFFSET, totalDataLen);
+ dataBuf.put(BIN_MSG_MSGTYPE_OFFSET, msgType);
+ dataBuf.putShort(BIN_MSG_GROUPIDNUM_OFFSET, (short) groupIdNum);
+ dataBuf.putShort(BIN_MSG_STREAMIDNUM_OFFSET, (short) streamIdNum);
+ dataBuf.putShort(BIN_MSG_EXTEND_OFFSET, (short) extendField);
+ dataBuf.putInt(BIN_MSG_DT_OFFSET, (int) dataTimeSec);
+ dataBuf.putShort(BIN_MSG_CNT_OFFSET, (short) msgCount);
+ dataBuf.putInt(BIN_MSG_UNIQ_OFFSET, (int) uniq);
+ dataBuf.putInt(BIN_MSG_BODYLEN_OFFSET, bodyData.length);
+ if (bodyData.length > 0) {
+ System.arraycopy(bodyData, 0, dataBuf.array(),
BIN_MSG_BODY_OFFSET, bodyData.length);
+ }
+ dataBuf.putShort(totalPkgLength
+ - BIN_MSG_ATTRLEN_SIZE - BIN_MSG_MAGIC_SIZE -
origAttr.length(), (short) origAttr.length());
+ if (origAttr.length() > 0) {
+ System.arraycopy(origAttr.getBytes(StandardCharsets.UTF_8), 0,
dataBuf.array(),
+ totalPkgLength - BIN_MSG_MAGIC_SIZE - origAttr.length(),
bodyData.length);
+ }
+ dataBuf.putShort(totalPkgLength - BIN_MSG_MAGIC_SIZE, (short)
BIN_MSG_MAGIC);
+ // build InLong message
+ InLongMsg inLongMsg = InLongMsg.newInLongMsg(source.isCompressed(), 4);
+ inLongMsg.addMsg(dataBuf.array());
+ long pkgTime = inLongMsg.getCreatetime();
+ Event event = EventBuilder.withBody(inLongMsg.buildArray(),
buildEventHeaders(pkgTime));
+ if (isOrderOrProxy) {
+ event = new SinkRspEvent(event, MsgType.MSG_BIN_MULTI_BODY,
channel);
+ }
+ inLongMsg.reset();
+ return event;
+ }
+
+ private boolean validAndFillTopic(BaseSource source) {
+ // valid groupId, streamId
+ ConfigManager configManager = ConfigManager.getInstance();
+ this.groupId = this.attrMap.get(AttributeConstants.GROUP_ID);
+ this.streamId = this.attrMap.get(AttributeConstants.STREAM_ID);
+ if (num2name) {
+ if (this.groupIdNum == 0) {
+ source.fileMetricEventInc(StatConstants.EVENT_WITHOUTGROUPID);
+ this.errCode = DataProxyErrCode.MISS_REQUIRED_GROUPID_ARGUMENT;
+ this.errMsg = "groupIdNum is 0 in message";
+ return false;
+ }
+ String confGroupId;
+ String confStreamId;
+ String strGroupIdNum = String.valueOf(this.groupIdNum);
+ if (configManager.getGroupIdMappingProperties() == null) {
+ source.fileMetricEventInc(StatConstants.EVENT_SERVICE_UNREADY);
+ this.errCode = DataProxyErrCode.CONF_SERVICE_UNREADY;
+ this.errMsg = "GroupId-Mapping configuration is null";
+ return false;
+ }
+ confGroupId =
configManager.getGroupIdMappingProperties().get(strGroupIdNum);
+ if (StringUtils.isBlank(confGroupId)) {
+ source.fileMetricEventInc(StatConstants.EVENT_WITHOUTGROUPID);
+ this.errCode =
DataProxyErrCode.GROUPID_OR_STREAMID_NOT_CONFIGURE;
+ this.errMsg = String.format("Non-existing groupIdNum(%s)
configuration", strGroupIdNum);
+ return false;
+ }
+ if (StringUtils.isNotBlank(this.groupId) &&
!this.groupId.equalsIgnoreCase(confGroupId)) {
+
source.fileMetricEventInc(StatConstants.EVENT_INCONSGROUPORSTREAMID);
+ this.errCode = DataProxyErrCode.GROUPID_OR_STREAMID_INCONSTANT;
+ this.errMsg = String.format(
+ "Inconstant GroupId not equal, (%s) in attr but (%s)
in configure by groupIdNum(%s)",
+ this.groupId, confGroupId, strGroupIdNum);
+ return false;
+ }
+ this.groupId = confGroupId;
+ // check streamId
+ if (this.streamIdNum == 0) {
+ if (StringUtils.isNotBlank(this.streamId)) {
+
source.fileMetricEventInc(StatConstants.EVENT_INCONSGROUPORSTREAMID);
+ this.errCode =
DataProxyErrCode.GROUPID_OR_STREAMID_INCONSTANT;
+ this.errMsg = String.format("Inconstant streamId(%s) in
attr but streamIdNum=0", this.streamId);
+ return false;
+ }
+ } else {
+ if (configManager.getStreamIdMappingProperties() == null) {
+
source.fileMetricEventInc(StatConstants.EVENT_SERVICE_UNREADY);
+ this.errCode = DataProxyErrCode.CONF_SERVICE_UNREADY;
+ this.errMsg = "StreamId-Mapping configuration is null";
+ return false;
+ }
+ Map<String, String> confStreamIdMap =
+
configManager.getStreamIdMappingProperties().get(strGroupIdNum);
+ if (confStreamIdMap == null) {
+
source.fileMetricEventInc(StatConstants.EVENT_SERVICE_UNREADY);
+ this.errCode = DataProxyErrCode.CONF_SERVICE_UNREADY;
+ this.errMsg = "GroupId in StreamId-Mapping configuration
is null";
+ return false;
+ }
+ String strStreamIdNum = String.valueOf(this.streamIdNum);
+ confStreamId = confStreamIdMap.get(strStreamIdNum);
+ if (StringUtils.isBlank(confStreamId)) {
+
source.fileMetricEventInc(StatConstants.EVENT_WITHOUTGROUPID);
+ this.errCode =
DataProxyErrCode.GROUPID_OR_STREAMID_NOT_CONFIGURE;
+ this.errMsg = String.format("Non-existing
GroupId(%s)-StreamId(%s) configuration",
+ strGroupIdNum, strStreamIdNum);
+ return false;
+ }
+ if (StringUtils.isNotBlank(this.streamId) &&
!this.streamId.equalsIgnoreCase(confStreamId)) {
+
source.fileMetricEventInc(StatConstants.EVENT_INCONSGROUPORSTREAMID);
+ this.errCode =
DataProxyErrCode.GROUPID_OR_STREAMID_INCONSTANT;
+ this.errMsg = String.format(
+ "Inconstant StreamId, (%s) in attr but (%s) in
configure by groupIdNum(%s), streamIdNum(%s)",
+ this.streamId, confStreamId, strGroupIdNum,
strStreamIdNum);
+ return false;
+ }
+ this.streamId = confStreamId;
+ }
+ // check whether enable num 2 name translate
+ String enableTrans =
(configManager.getGroupIdEnableMappingProperties() == null)
+ ? null
+ :
configManager.getGroupIdEnableMappingProperties().get(strGroupIdNum);
+ if ("true".equalsIgnoreCase(enableTrans) && this.num2name) {
+ this.transNum2Name = true;
+ }
+ } else {
+ if (StringUtils.isBlank(groupId)) {
+ source.fileMetricEventInc(StatConstants.EVENT_WITHOUTGROUPID);
+ this.errCode = DataProxyErrCode.MISS_REQUIRED_GROUPID_ARGUMENT;
+ return false;
+ }
+ }
+ // get and check topic configure
+ this.topicName = configManager.getTopicName(this.groupId,
this.streamId);
+ if (StringUtils.isBlank(this.topicName)) {
+ if (CommonConfigHolder.getInstance().isNoTopicAccept()) {
+ this.topicName = source.getDefTopic();
+ } else {
+ source.fileMetricEventInc(StatConstants.EVENT_NOTOPIC);
+ this.errCode = DataProxyErrCode.TOPIC_IS_BLANK;
+ this.errMsg = String.format("Topic is null for
inlongGroupId=(%s), inlongStreamId=(%s)",
+ this.groupId, this.streamId);
+ return false;
+ }
+ }
+ if (StringUtils.isBlank(this.streamId)) {
+ this.streamId = "";
+ }
+ return true;
+ }
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/CodecTextMsg.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/CodecTextMsg.java
new file mode 100644
index 000000000..9d521ccb1
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/CodecTextMsg.java
@@ -0,0 +1,257 @@
+/*
+ * 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.dataproxy.source2.v0msg;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flume.Event;
+import org.apache.flume.event.EventBuilder;
+import org.apache.inlong.common.enums.DataProxyErrCode;
+import org.apache.inlong.common.msg.AttributeConstants;
+import org.apache.inlong.common.msg.InLongMsg;
+import org.apache.inlong.common.msg.MsgType;
+import org.apache.inlong.dataproxy.config.CommonConfigHolder;
+import org.apache.inlong.dataproxy.config.ConfigManager;
+import org.apache.inlong.dataproxy.consts.StatConstants;
+import org.apache.inlong.dataproxy.source2.BaseSource;
+import org.xerial.snappy.Snappy;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.TXT_MSG_BODYLEN_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.TXT_MSG_BODY_OFFSET;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.TXT_MSG_FORMAT_SIZE;
+import static
org.apache.inlong.dataproxy.source2.v0msg.MsgFieldConsts.TXT_MSG_TOTALLEN_SIZE;
+
+public class CodecTextMsg extends AbsV0MsgCodec {
+
+ public CodecTextMsg(int totalDataLen, int msgTypeValue,
+ long msgRcvTime, String strRemoteIP) {
+ super(totalDataLen, msgTypeValue, msgRcvTime, strRemoteIP);
+ }
+
+ public boolean descMsg(BaseSource source, ByteBuf cb) throws Exception {
+ // get body length
+ int msgHeadPos = cb.readerIndex() - 5;
+ int bodyLen = cb.getInt(msgHeadPos + TXT_MSG_BODYLEN_OFFSET);
+ if (bodyLen <= 0) {
+ if (bodyLen == 0) {
+ source.fileMetricEventInc(StatConstants.EVENT_NOBODY);
+ this.errCode = DataProxyErrCode.BODY_LENGTH_ZERO;
+ } else {
+ source.fileMetricEventInc(StatConstants.EVENT_NEGBODY);
+ this.errCode = DataProxyErrCode.BODY_LENGTH_LESS_ZERO;
+ }
+ return false;
+ }
+ if (bodyLen + TXT_MSG_FORMAT_SIZE > totalDataLen +
TXT_MSG_TOTALLEN_SIZE) {
+ this.errCode = DataProxyErrCode.BODY_EXCEED_MAX_LEN;
+ this.errMsg = String.format("Error msg, bodyLen(%d) +
fixedLength(%d) > totalDataLen(%d) + 4",
+ bodyLen, TXT_MSG_FORMAT_SIZE, totalDataLen);
+ return false;
+ }
+ // extract body bytes
+ this.bodyData = new byte[bodyLen];
+ cb.getBytes(msgHeadPos + TXT_MSG_BODY_OFFSET, this.bodyData, 0,
bodyLen);
+ // get attribute length
+ int attrLen = cb.getInt(msgHeadPos + TXT_MSG_BODY_OFFSET + bodyLen);
+ if (attrLen < 0) {
+ this.errCode = DataProxyErrCode.ATTR_LENGTH_LESS_ZERO;
+ return false;
+ }
+ // check attribute length
+ if (totalDataLen + TXT_MSG_TOTALLEN_SIZE != TXT_MSG_FORMAT_SIZE +
bodyLen + attrLen) {
+ this.errCode = DataProxyErrCode.BODY_EXCEED_MAX_LEN;
+ this.errMsg = String.format(
+ "Error msg, totalDataLen(%d) + 4 != fixedLength(%d) +
bodyLen(%d) + attrLen(%d)",
+ totalDataLen, TXT_MSG_FORMAT_SIZE, bodyLen, attrLen);
+ return false;
+ }
+ // extract attr bytes
+ if (!decAttrInfo(source, cb, attrLen, msgHeadPos + TXT_MSG_FORMAT_SIZE
+ bodyLen)) {
+ return false;
+ }
+ // decompress body data
+ if
(StringUtils.isNotBlank(attrMap.get(AttributeConstants.COMPRESS_TYPE))) {
+ byte[] unCompressedData;
+ try {
+ int uncompressedLen = Snappy.uncompressedLength(bodyData, 0,
bodyData.length);
+ unCompressedData = new byte[uncompressedLen];
+ Snappy.uncompress(bodyData, 0, bodyData.length,
unCompressedData, 0);
+ } catch (IOException e) {
+ source.fileMetricEventInc(StatConstants.EVENT_UNPRESSEXP);
+ this.errCode = DataProxyErrCode.UNCOMPRESS_DATA_ERROR;
+ this.errMsg = String.format("Error to uncompress msg, compress
type(%s), attr: (%s), error: (%s)",
+ attrMap.get(AttributeConstants.COMPRESS_TYPE),
origAttr, e.getCause());
+ return false;
+ }
+ if (unCompressedData.length == 0) {
+ source.fileMetricEventInc(StatConstants.EVENT_UNPRESSEXP);
+ this.errCode = DataProxyErrCode.UNCOMPRESS_DATA_ERROR;
+ this.errMsg = String.format("Error to uncompress msg, compress
type(%s), attr: (%s), error: 2",
+ attrMap.get(AttributeConstants.COMPRESS_TYPE),
origAttr);
+ return false;
+ }
+ this.bodyData = unCompressedData;
+ }
+ // check body items
+ if (MsgType.MSG_MULTI_BODY.equals(MsgType.valueOf(msgType))) {
+ int readPos = 0;
+ int singleMsgLen;
+ ByteBuffer bodyBuffer = ByteBuffer.wrap(bodyData);
+ while (bodyBuffer.remaining() > 0) {
+ singleMsgLen = bodyBuffer.getInt(readPos);
+ if (singleMsgLen <= 0 || singleMsgLen >
bodyBuffer.remaining()) {
+ source.fileMetricEventInc(StatConstants.EVENT_MALFORMED);
+ this.errCode = DataProxyErrCode.BODY_EXCEED_MAX_LEN;
+ this.errMsg = String.format(
+ "Malformed data len, singleMsgLen(%d), buffer
remaining(%d), attr: (%s)",
+ singleMsgLen, bodyBuffer.remaining(), origAttr);
+ return false;
+ }
+ readPos += 4 + singleMsgLen;
+ }
+ }
+ return true;
+ }
+
+ public boolean validAndFillFields(BaseSource source, StringBuilder
strBuff) {
+ // process topic field
+ String tmpGroupId = attrMap.get(AttributeConstants.GROUP_ID);
+ String tmpStreamId = attrMap.get(AttributeConstants.STREAM_ID);
+ if (StringUtils.isBlank(tmpGroupId)) {
+ source.fileMetricEventInc(StatConstants.EVENT_WITHOUTGROUPID);
+ this.errCode = DataProxyErrCode.MISS_REQUIRED_GROUPID_ARGUMENT;
+ return false;
+ }
+ // get and check topic configure
+ String tmpTopicName =
ConfigManager.getInstance().getTopicName(tmpGroupId, tmpStreamId);
+ if (StringUtils.isBlank(tmpTopicName)) {
+ if (CommonConfigHolder.getInstance().isNoTopicAccept()) {
+ tmpTopicName = source.getDefTopic();
+ } else {
+ source.fileMetricEventInc(StatConstants.EVENT_NOTOPIC);
+ this.errCode = DataProxyErrCode.TOPIC_IS_BLANK;
+ this.errMsg = String.format(
+ "Topic is null for inlongGroupId=(%s),
inlongStreamId=(%s)", tmpGroupId, tmpStreamId);
+ return false;
+ }
+ }
+ this.groupId = tmpGroupId;
+ this.topicName = tmpTopicName;
+ if (StringUtils.isNotBlank(tmpStreamId)) {
+ this.streamId = tmpStreamId;
+ }
+ // process message count
+ this.msgCount = 1;
+ String cntStr = attrMap.get(AttributeConstants.MESSAGE_COUNT);
+ if (StringUtils.isBlank(cntStr)) {
+ attrMap.put(AttributeConstants.MESSAGE_COUNT,
String.valueOf(this.msgCount));
+ } else {
+ try {
+ this.msgCount = Integer.parseInt(cntStr);
+ } catch (Throwable e) {
+ attrMap.put(AttributeConstants.MESSAGE_COUNT,
String.valueOf(this.msgCount));
+ }
+ }
+ // process data-time
+ this.dataTimeMs = msgRcvTime;
+ String strDataTime = attrMap.get(AttributeConstants.DATA_TIME);
+ if (StringUtils.isBlank(strDataTime)) {
+ attrMap.put(AttributeConstants.DATA_TIME,
String.valueOf(this.dataTimeMs));
+ } else {
+ try {
+ this.dataTimeMs = Long.parseLong(strDataTime);
+ } catch (Throwable e) {
+ attrMap.put(AttributeConstants.DATA_TIME,
String.valueOf(this.dataTimeMs));
+ }
+ }
+ // process sequence id
+ String sequenceId = attrMap.get(AttributeConstants.SEQUENCE_ID);
+ if (StringUtils.isNotBlank(sequenceId)) {
+
strBuff.append(topicName).append(AttributeConstants.SEPARATOR).append(streamId)
+ .append(AttributeConstants.SEPARATOR).append(sequenceId)
+ .append("#").append(strRemoteIP);
+ msgSeqId = strBuff.toString();
+ strBuff.delete(0, strBuff.length());
+ }
+ // append required attributes
+ if (StringUtils.isBlank(attrMap.get(AttributeConstants.RCV_TIME))) {
+ strBuff.append(AttributeConstants.RCV_TIME)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(msgRcvTime);
+ attrMap.put(AttributeConstants.RCV_TIME,
String.valueOf(msgRcvTime));
+ }
+ if (StringUtils.isBlank(attrMap.get(AttributeConstants.MSG_RPT_TIME)))
{
+ if (strBuff.length() > 0) {
+ strBuff.append(AttributeConstants.SEPARATOR);
+ }
+ strBuff.append(AttributeConstants.MSG_RPT_TIME)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(msgRcvTime);
+ attrMap.put(AttributeConstants.MSG_RPT_TIME,
String.valueOf(msgRcvTime));
+ }
+ // rebuild attribute string
+ if (strBuff.length() > 0) {
+ if (StringUtils.isNotBlank(origAttr)) {
+ strBuff.append(AttributeConstants.SEPARATOR).append(origAttr);
+ }
+ totalDataLen += strBuff.length() - origAttr.length();
+ origAttr = strBuff.toString();
+ strBuff.delete(0, strBuff.length());
+ }
+ return true;
+ }
+
+ public Event encEventPackage(BaseSource source, Channel channel) {
+ // build InLongMsg object
+ int inLongMsgVer = 1;
+ if (MsgType.MSG_MULTI_BODY_ATTR.equals(MsgType.valueOf(msgType))) {
+ inLongMsgVer = 3;
+ }
+ InLongMsg inLongMsg = InLongMsg.newInLongMsg(source.isCompressed(),
inLongMsgVer);
+ if (MsgType.MSG_MULTI_BODY.equals(MsgType.valueOf(msgType))) {
+ int calcCnt = 0;
+ int singleMsgLen;
+ ByteBuffer bodyBuffer = ByteBuffer.wrap(bodyData);
+ attrMap.put(AttributeConstants.MESSAGE_COUNT, String.valueOf(1));
+ while (bodyBuffer.remaining() > 0) {
+ singleMsgLen = bodyBuffer.getInt();
+ if (singleMsgLen <= 0 || singleMsgLen >
bodyBuffer.remaining()) {
+ break;
+ }
+ byte[] record = new byte[singleMsgLen];
+ bodyBuffer.get(record);
+ inLongMsg.addMsg(mapJoiner.join(attrMap), bodyBuffer);
+ calcCnt++;
+ }
+ attrMap.put(AttributeConstants.MESSAGE_COUNT,
String.valueOf(calcCnt));
+ this.msgCount = calcCnt;
+ } else if
(MsgType.MSG_MULTI_BODY_ATTR.equals(MsgType.valueOf(msgType))) {
+ attrMap.put(AttributeConstants.MESSAGE_COUNT, String.valueOf(1));
+ inLongMsg.addMsg(mapJoiner.join(attrMap), bodyData);
+ attrMap.put(AttributeConstants.MESSAGE_COUNT,
String.valueOf(this.msgCount));
+ } else {
+ inLongMsg.addMsg(mapJoiner.join(attrMap), bodyData);
+ }
+ long pkgTime = inLongMsg.getCreatetime();
+ Event event = EventBuilder.withBody(inLongMsg.buildArray(),
buildEventHeaders(pkgTime));
+ inLongMsg.reset();
+ return event;
+ }
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/MsgFieldConsts.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/MsgFieldConsts.java
new file mode 100644
index 000000000..20b0e884e
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/MsgFieldConsts.java
@@ -0,0 +1,74 @@
+/*
+ * 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.dataproxy.source2.v0msg;
+
+public class MsgFieldConsts {
+
+ public static final int BIN_MSG_FORMAT_SIZE = 29;
+ public static final int BIN_MSG_TOTALLEN_OFFSET = 0;
+ public static final int BIN_MSG_TOTALLEN_SIZE = 4;
+ public static final int BIN_MSG_FIXED_CONTENT_SIZE = BIN_MSG_FORMAT_SIZE -
BIN_MSG_TOTALLEN_SIZE;
+ public static final int BIN_MSG_MSGTYPE_OFFSET = 4;
+ public static final int BIN_MSG_MSGTYPE_SIZE = 1;
+ public static final int BIN_MSG_GROUPIDNUM_OFFSET = 5;
+ public static final int BIN_MSG_GROUPIDNUM_SIZE = 2;
+ public static final int BIN_MSG_STREAMIDNUM_OFFSET = 7;
+ public static final int BIN_MSG_STREAMIDNUM_SIZE = 2;
+ public static final int BIN_MSG_EXTEND_OFFSET = 9;
+ public static final int BIN_MSG_EXTEND_SIZE = 2;
+ public static final int BIN_MSG_DT_OFFSET = 11;
+ public static final int BIN_MSG_DT_SIZE = 4;
+ public static final int BIN_MSG_CNT_OFFSET = 15;
+ public static final int BIN_MSG_CNT_SIZE = 2;
+ public static final int BIN_MSG_UNIQ_OFFSET = 17;
+ public static final int BIN_MSG_UNIQ_SIZE = 4;
+ public static final int BIN_MSG_SET_SNAPPY = (1 << 5);
+ public static final int BIN_MSG_BODYLEN_OFFSET = 21;
+ public static final int BIN_MSG_BODYLEN_SIZE = 4;
+ public static final int BIN_MSG_BODY_OFFSET = BIN_MSG_BODYLEN_SIZE +
BIN_MSG_BODYLEN_OFFSET;
+ public static final int BIN_MSG_ATTRLEN_SIZE = 2;
+ public static final int BIN_MSG_MAGIC_SIZE = 2;
+ public static final int BIN_MSG_MAGIC = 0xEE01;
+
+ public static final int BIN_HB_FORMAT_SIZE = 18;
+ public static final int BIN_HB_TOTALLEN_OFFSET = 0;
+ public static final int BIN_HB_TOTALLEN_SIZE = 4;
+ public static final int BIN_HB_FIXED_CONTENT_SIZE = BIN_HB_FORMAT_SIZE -
BIN_HB_TOTALLEN_SIZE;
+ public static final int BIN_HB_MSGTYPE_OFFSET = 4;
+ public static final int BIN_HB_MSGTYPE_SIZE = 1;
+ public static final int BIN_HB_DATATIME_OFFSET = 5;
+ public static final int BIN_HB_DATATIME_SIZE = 4;
+ public static final int BIN_HB_VERSION_OFFSET = 9;
+ public static final int BIN_HB_VERSION_SIZE = 1;
+ public static final int BIN_HB_BODYLEN_OFFSET = 10;
+ public static final int BIN_HB_BODYLEN_SIZE = 4;
+ public static final int BIN_HB_BODY_OFFSET = BIN_HB_BODYLEN_SIZE +
BIN_HB_BODYLEN_OFFSET;
+ public static final int BIN_HB_ATTRLEN_SIZE = 2;
+
+ public static final int TXT_MSG_FORMAT_SIZE = 13;
+ public static final int TXT_MSG_TOTALLEN_OFFSET = 0;
+ public static final int TXT_MSG_TOTALLEN_SIZE = 4;
+ public static final int TXT_MSG_FIXED_CONTENT_SIZE = TXT_MSG_FORMAT_SIZE -
TXT_MSG_TOTALLEN_SIZE;
+ public static final int TXT_MSG_MSGTYPE_OFFSET = 4;
+ public static final int TXT_MSG_MSGTYPE_SIZE = 1;
+ public static final int TXT_MSG_BODYLEN_OFFSET = 5;
+ public static final int TXT_MSG_BODYLEN_SIZE = 4;
+ public static final int TXT_MSG_BODY_OFFSET = TXT_MSG_BODYLEN_SIZE +
TXT_MSG_BODYLEN_OFFSET;
+ public static final int TXT_MSG_ATTRLEN_SIZE = 4;
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/AddressUtils.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/AddressUtils.java
index 5ed9eb345..073341370 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/AddressUtils.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/AddressUtils.java
@@ -26,20 +26,32 @@ public class AddressUtils {
private static final Logger logger =
LoggerFactory.getLogger(AddressUtils.class);
+ public static String getChannelLocalIP(Channel channel) {
+ return getChannelIP(channel, true);
+ }
+
public static String getChannelRemoteIP(Channel channel) {
+ return getChannelIP(channel, false);
+ }
+
+ private static String getChannelIP(Channel channel, boolean isLocal) {
if (channel == null) {
return null;
}
- SocketAddress rmtAddress = channel.remoteAddress();
- if (rmtAddress == null) {
+ SocketAddress address = isLocal ? channel.localAddress() :
channel.remoteAddress();
+ if (address == null) {
return null;
}
- String strRemoteIP = rmtAddress.toString();
+ String strAddrIP = address.toString();
try {
- strRemoteIP = strRemoteIP.substring(1, strRemoteIP.indexOf(':'));
- return strRemoteIP;
+ strAddrIP = strAddrIP.substring(1, strAddrIP.indexOf(':'));
+ return strAddrIP;
} catch (Exception ee) {
- logger.warn("Fail to get the remote IP, rmtAddress = {}",
rmtAddress);
+ if (isLocal) {
+ logger.warn("Fail to get the local IP, localAddress = {}",
address);
+ } else {
+ logger.warn("Fail to get the remote IP, remoteAddress = {}",
address);
+ }
return null;
}
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java
index fdb3f8d9f..74ed688ee 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java
@@ -22,7 +22,6 @@ import static
org.apache.inlong.common.util.NetworkUtils.getLocalIp;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandlerContext;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
@@ -195,11 +194,10 @@ public class MessageUtils {
ByteBuf binBuffer;
final StringBuilder strBuff = new StringBuilder(512);
// get and check channel context
- ChannelHandlerContext ctx = event.getCtx();
- if (ctx == null || ctx.channel() == null || !ctx.channel().isActive())
{
+ Channel remoteChannel = event.getChannel();
+ if (remoteChannel == null || !remoteChannel.isActive()) {
return;
}
- Channel remoteChannel = ctx.channel();
// check message type
MsgType msgType = event.getMsgType();
if (MsgType.MSG_UNKNOWN.equals(msgType)
