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 68c41dd5b [INLONG-8073][DataProxy] Add HTTP message processing logic
in source2 (#8074)
68c41dd5b is described below
commit 68c41dd5b4a1acc2ad4667c222c93fc9f9c87d14
Author: Goson Zhang <[email protected]>
AuthorDate: Wed May 24 09:35:17 2023 +0800
[INLONG-8073][DataProxy] Add HTTP message processing logic in source2
(#8074)
---
.../inlong/dataproxy/consts/StatConstants.java | 25 +-
.../inlong/dataproxy/source2/BaseSource.java | 162 +++++----
.../dataproxy/source2/InLongMessageFactory.java | 17 +-
.../dataproxy/source2/InLongMessageHandler.java | 50 +--
...{SimpleTcpSource.java => SimpleHttpSource.java} | 77 +----
.../inlong/dataproxy/source2/SimpleTcpSource.java | 18 +-
.../inlong/dataproxy/source2/SimpleUdpSource.java | 2 +
.../inlong/dataproxy/source2/SourceConstants.java | 3 +-
.../source2/httpMsg/InLongHttpMsgHandler.java | 385 +++++++++++++++++++++
.../dataproxy/source2/v0msg/AbsV0MsgCodec.java | 4 +-
.../dataproxy/source2/v0msg/CodecBinMsg.java | 4 +-
.../dataproxy/source2/v0msg/CodecTextMsg.java | 2 +
.../source2/v1msg/InlongTcpSourceCallback.java | 116 +++++++
.../inlong/dataproxy/utils/ConfStringUtils.java | 56 ++-
14 files changed, 717 insertions(+), 204 deletions(-)
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 07b24c26b..6dbc4f14b 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
@@ -19,6 +19,11 @@ package org.apache.inlong.dataproxy.consts;
public class StatConstants {
+ 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_VISITIP_ILLEGAL =
"links.illegal";
+ public static final java.lang.String EVENT_NOTOPIC = "config.notopic";
+
public static final java.lang.String METASINK_SUCCESS = "metasink.success";
public static final java.lang.String METASINK_DROPPED = "metasink.dropped";
public static final java.lang.String METASINK_RETRY = "metasink.retry";
@@ -29,8 +34,8 @@ public class StatConstants {
public static final java.lang.String METASINK_PROCESS_SPEED =
"metasink.process.speed";
public static final java.lang.String EVENT_OTHEREXP = "socketmsg.otherexp";
public static final java.lang.String EVENT_INVALID = "socketmsg.invalid";
+ // source
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";
@@ -49,11 +54,23 @@ public class StatConstants {
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";
+ // http
+ public static final java.lang.String EVENT_HTTP_DECFAIL =
"httpmsg.decfailure";
+ public static final java.lang.String EVENT_HTTP_INVALIDMETHOD =
"httpmsg.invmethod";
+ public static final java.lang.String EVENT_HTTP_BLANKURI =
"httpmsg.blankuri";
+ public static final java.lang.String EVENT_HTTP_URIDECFAIL =
"httpmsg.decurifail";
+ public static final java.lang.String EVENT_HTTP_INVALIDURI =
"httpmsg.invuri";
+ public static final java.lang.String EVENT_HTTP_ILLEGAL_VISIT =
"httpmsg.illegal";
+ public static final java.lang.String EVENT_HTTP_HB_SUCCESS =
"httphb.success";
+ public static final java.lang.String EVENT_HTTP_WITHOUTGROUPID =
"httpmsg.wogroupid";
+ public static final java.lang.String EVENT_HTTP_WITHOUTSTREAMID =
"httpmsg.wostreamid";
+ public static final java.lang.String EVENT_HTTP_NOBODY = "httpmsg.nobody";
+ public static final java.lang.String EVENT_HTTP_EMPTYBODY =
"httpmsg.emptybody";
+ public static final java.lang.String EVENT_HTTP_BODYOVERMAXLEN =
"httpmsg.bodyovermax";
+ public static final java.lang.String EVENT_HTTP_POST_SUCCESS =
"httpmsg.success";
+ public static final java.lang.String EVENT_HTTP_POST_DROPPED =
"httpmsg.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";
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
index 6fd784681..d843f8b1e 100644
---
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
@@ -18,15 +18,11 @@
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.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.FlumeException;
import org.apache.flume.conf.Configurable;
@@ -37,7 +33,10 @@ 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.DataProxyMetricItem;
import org.apache.inlong.dataproxy.metrics.DataProxyMetricItemSet;
+import org.apache.inlong.dataproxy.metrics.audit.AuditUtils;
+import org.apache.inlong.dataproxy.source2.httpMsg.InLongHttpMsgHandler;
import org.apache.inlong.dataproxy.utils.ConfStringUtils;
import org.apache.inlong.dataproxy.utils.FailoverChannelProcessorHolder;
import org.apache.inlong.sdk.commons.admin.AdminServiceRegister;
@@ -45,11 +44,18 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Constructor;
+import java.util.HashMap;
import java.util.Map;
+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;
+
/**
* source base class
- *
*/
public abstract class BaseSource
extends
@@ -107,9 +113,9 @@ public abstract class BaseSource
protected int monitorStatInvlSec;
protected int maxMonitorStatCnt;
protected MonitorIndex monitorIndex = null;
- private MonitorIndexExt monitorIndexExt = null;
// metric set
protected DataProxyMetricItemSet metricItemSet;
+ private MonitorIndexExt monitorIndexExt = null;
public BaseSource() {
super();
@@ -129,8 +135,12 @@ public abstract class BaseSource
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());
+ tmpVal =
context.getString(SourceConstants.SRCCXT_MESSAGE_HANDLER_NAME);
+ if (StringUtils.isBlank(tmpVal)) {
+ tmpVal =
SourceConstants.SRC_PROTOCOL_TYPE_HTTP.equalsIgnoreCase(getProtocolName())
+ ? InLongHttpMsgHandler.class.getName()
+ : InLongMessageHandler.class.getName();
+ }
Preconditions.checkArgument(StringUtils.isNotBlank(tmpVal),
SourceConstants.SRCCXT_MESSAGE_HANDLER_NAME + " config is
blank");
this.messageHandlerName = tmpVal;
@@ -145,8 +155,8 @@ public abstract class BaseSource
this.defAttr = tmpVal.trim();
}
// get allowed max message length
- this.maxMsgLength = getIntValue(context,
SourceConstants.SRCCXT_MAX_MSG_LENGTH,
- SourceConstants.VAL_DEF_MAX_MSG_LENGTH);
+ this.maxMsgLength = ConfStringUtils.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 ["
@@ -162,42 +172,44 @@ public abstract class BaseSource
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);
+ this.maxAcceptThreads = ConfStringUtils.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);
+ this.maxWorkerThreads = ConfStringUtils.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);
+ this.maxReadIdleTimeMs = ConfStringUtils.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
+ && this.maxReadIdleTimeMs <=
SourceConstants.VAL_MAX_READ_IDLE_TIME_MS),
+ SourceConstants.SRCCXT_MAX_READ_IDLE_TIME_MS + " must be in ["
+ + SourceConstants.VAL_MIN_READ_IDLE_TIME_MS + ", "
+ + SourceConstants.VAL_MAX_READ_IDLE_TIME_MS + "]");
// get file metric statistic
- this.monitorStatInvlSec = getIntValue(context,
SourceConstants.SRCCXT_STAT_INTERVAL_SEC,
- SourceConstants.VAL_DEF_STAT_INVL_SEC);
+ this.monitorStatInvlSec = ConfStringUtils.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);
+ this.maxMonitorStatCnt = ConfStringUtils.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);
+ this.maxConnections = ConfStringUtils.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);
@@ -205,8 +217,8 @@ public abstract class BaseSource
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);
+ this.maxRcvBufferSize = ConfStringUtils.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);
@@ -214,8 +226,8 @@ public abstract class BaseSource
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);
+ this.maxSendBufferSize = ConfStringUtils.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);
@@ -281,11 +293,19 @@ public abstract class BaseSource
monitorIndexExt.shutDown();
}
}
+ // stop workers
+ if (this.acceptorGroup != null) {
+ this.acceptorGroup.shutdownGracefully();
+ }
+ if (this.workerGroup != null) {
+ this.workerGroup.shutdownGracefully();
+ }
logger.info("[STOP {} SOURCE]{} stopped", this.getProtocolName(),
this.getName());
}
/**
* get metricItemSet
+ *
* @return the metricItemSet
*/
public DataProxyMetricItemSet getMetricItemSet() {
@@ -364,8 +384,34 @@ public abstract class BaseSource
}
}
+ /**
+ * addMetric
+ *
+ * @param result
+ * @param size
+ * @param event
+ */
+ public 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, getName());
+ dimensions.put(DataProxyMetricItem.KEY_SOURCE_DATA_ID, getStrPort());
+ DataProxyMetricItem.fillInlongId(event, dimensions);
+ DataProxyMetricItem.fillAuditFormatTime(event, dimensions);
+ DataProxyMetricItem metricItem =
metricItemSet.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);
+ }
+ }
+
/**
* channel factory
+ *
* @return
*/
public ChannelInitializer getChannelInitializerFactory() {
@@ -415,60 +461,10 @@ public abstract class BaseSource
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
+ * @param context
* @return
*/
private String getHostIp(Context context) {
@@ -498,7 +494,7 @@ public abstract class BaseSource
/**
* getHostPort
*
- * @param context
+ * @param context
* @return
*/
private int getHostPort(Context context) {
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
index 2dfe66df8..1b2a604fe 100644
---
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
@@ -19,25 +19,27 @@ 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.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
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;
+ private static final Logger LOG =
LoggerFactory.getLogger(InLongMessageFactory.class);
+ private final BaseSource source;
/**
* get server factory
@@ -58,6 +60,13 @@ public class InLongMessageFactory extends
ChannelInitializer<SocketChannel> {
INLONG_LENGTH_ADJUSTMENT, INLONG_INITIAL_BYTES_TO_STRIP,
DEFAULT_FAIL_FAST));
ch.pipeline().addLast("readTimeoutHandler",
new ReadTimeoutHandler(source.getMaxReadIdleTimeMs(),
TimeUnit.MILLISECONDS));
+ } else if
(source.getProtocolName().equalsIgnoreCase(SourceConstants.SRC_PROTOCOL_TYPE_HTTP))
{
+ // add http message codec
+ ch.pipeline().addLast("msgCodec", new HttpServerCodec());
+ ch.pipeline().addLast("msgAggregator", new
HttpObjectAggregator(source.getMaxMsgLength()));
+ ch.pipeline().addLast("readTimeoutHandler",
+ new ReadTimeoutHandler(source.getMaxReadIdleTimeMs(),
TimeUnit.MILLISECONDS));
+
}
// build message handler
if (source.getChannelProcessor() != null) {
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
index e92a58b98..ffe0c65c6 100644
---
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
@@ -29,12 +29,10 @@ 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.source2.v1msg.InlongTcpSourceCallback;
import org.apache.inlong.dataproxy.utils.AddressUtils;
import org.apache.inlong.dataproxy.utils.DateTimeUtils;
import org.apache.inlong.sdk.commons.protocol.EventUtils;
@@ -45,10 +43,9 @@ 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;
@@ -209,7 +206,7 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
String strRemoteIp =
AddressUtils.getChannelRemoteIP(ctx.channel());
if (strRemoteIp != null
&& ConfigManager.getInstance().isIllegalIP(strRemoteIp)) {
- source.fileMetricEventInc(StatConstants.EVENT_LINKS_ILLEGAL);
+ source.fileMetricEventInc(StatConstants.EVENT_VISITIP_ILLEGAL);
ctx.channel().disconnect();
ctx.channel().close();
logger.error(strRemoteIp + " is Illegal IP, so refuse it !");
@@ -294,13 +291,13 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
source.fileMetricEventInc(StatConstants.EVENT_POST_SUCCESS);
source.fileMetricRecordAdd(strBuff.toString(),
msgCodec.getMsgCount(), 1, msgCodec.getBodyLength(), 0);
- this.addMetric(true, event.getBody().length, event);
+ source.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);
+ source.addMetric(false, event.getBody().length, event);
strBuff.delete(0, strBuff.length());
throw new ChannelException("ProcessEvent error can't write event
to channel.");
}
@@ -314,7 +311,7 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
ProxySdk.MessagePack packObject =
ProxySdk.MessagePack.parseFrom(msgBytes);
// reject service
if (source.isRejectService()) {
- this.addMetric(false, 0, null);
+ source.addMetric(false, 0, null);
source.fileMetricEventInc(StatConstants.EVENT_SERVICE_CLOSED);
this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT,
packObject);
return;
@@ -381,7 +378,7 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
try {
source.getChannelProcessor().processEvent(packEvent);
events.forEach(event -> {
- this.addMetric(true, event.getBody().length, event);
+ source.addMetric(true, event.getBody().length, event);
source.fileMetricEventInc(StatConstants.EVENT_POST_SUCCESS);
});
boolean awaitResult = callback.getLatch().await(
@@ -394,7 +391,7 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
} 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.addMetric(false, event.getBody().length, event);
source.fileMetricEventInc(StatConstants.EVENT_POST_DROPPED);
});
if (!callback.getHasResponsed().getAndSet(true)) {
@@ -421,7 +418,7 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
topic = source.getDefTopic();
} else {
source.fileMetricEventInc(StatConstants.EVENT_NOTOPIC);
- this.addMetric(false, event.getBody().length, event);
+ source.addMetric(false, event.getBody().length, event);
this.responsePackage(ctx,
ProxySdk.ResultCode.ERR_ID_ERROR, packObject);
return;
}
@@ -430,11 +427,11 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
// put to channel
try {
source.getChannelProcessor().processEvent(event);
- this.addMetric(true, event.getBody().length, event);
+ source.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);
+ source.addMetric(false, event.getBody().length, event);
this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT,
packObject);
source.fileMetricEventInc(StatConstants.EVENT_POST_DROPPED);
return;
@@ -653,29 +650,4 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
}
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/SimpleHttpSource.java
similarity index 59%
copy from
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SimpleTcpSource.java
copy to
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/SimpleHttpSource.java
index 6be918b0e..89ad9f335 100644
---
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/SimpleHttpSource.java
@@ -18,40 +18,36 @@
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.apache.inlong.dataproxy.utils.ConfStringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
-import java.util.Iterator;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.util.concurrent.DefaultThreadFactory;
/**
* Simple tcp source
- *
*/
-public class SimpleTcpSource extends BaseSource implements Configurable,
ConfigUpdateCallback {
+public class SimpleHttpSource extends BaseSource implements Configurable {
- private static final Logger logger =
LoggerFactory.getLogger(SimpleTcpSource.class);
+ private static final Logger logger =
LoggerFactory.getLogger(SimpleHttpSource.class);
private ServerBootstrap bootstrap;
private boolean tcpNoDelay;
- private boolean tcpKeepAlive;
private int highWaterMark;
- private boolean enableBusyWait;
- public SimpleTcpSource() {
+ public SimpleHttpSource() {
super();
- ConfigManager.getInstance().regIPVisitConfigChgCallback(this);
}
@Override
@@ -61,15 +57,9 @@ public class SimpleTcpSource extends BaseSource implements
Configurable, ConfigU
// 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);
+ this.highWaterMark = ConfStringUtils.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);
@@ -79,21 +69,19 @@ public class SimpleTcpSource extends BaseSource implements
Configurable, ConfigU
public synchronized void startSource() {
logger.info("start " + this.getName());
// build accept group
- this.acceptorGroup = EventLoopUtil.newEventLoopGroup(maxAcceptThreads,
enableBusyWait,
+ this.acceptorGroup = new NioEventLoopGroup(maxAcceptThreads,
new DefaultThreadFactory(this.getName() + "-boss-group"));
// build worker group
- this.workerGroup = EventLoopUtil.newEventLoopGroup(maxWorkerThreads,
enableBusyWait,
+ this.workerGroup = new NioEventLoopGroup(maxWorkerThreads,
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.channel(NioServerSocketChannel.class);
bootstrap.group(acceptorGroup, workerGroup);
bootstrap.childHandler(this.getChannelInitializerFactory());
try {
@@ -119,35 +107,6 @@ public class SimpleTcpSource extends BaseSource implements
Configurable, ConfigU
@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));
- }
+ return SourceConstants.SRC_PROTOCOL_TYPE_HTTP;
}
-
}
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
index 6be918b0e..6a14778c5 100644
---
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
@@ -18,16 +18,13 @@
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.ConfStringUtils;
import org.apache.inlong.dataproxy.utils.EventLoopUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,9 +32,14 @@ import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.util.Iterator;
+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;
+
/**
* Simple tcp source
- *
*/
public class SimpleTcpSource extends BaseSource implements Configurable,
ConfigUpdateCallback {
@@ -68,8 +70,8 @@ public class SimpleTcpSource extends BaseSource implements
Configurable, ConfigU
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);
+ this.highWaterMark = ConfStringUtils.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);
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
index dbe988f68..f8dfd0b18 100644
---
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
@@ -22,7 +22,9 @@ 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;
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
index edb111dd8..1f556c895 100644
---
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
@@ -102,8 +102,9 @@ public class SourceConstants {
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_DEF_READ_IDLE_TIME_MS = 3 * 60 * 1000;
public static final long VAL_MIN_READ_IDLE_TIME_MS = 60 * 1000;
+ public static final long VAL_MAX_READ_IDLE_TIME_MS = 70 * 60 * 1000;
// source protocol type
public static final String SRC_PROTOCOL_TYPE_TCP = "tcp";
public static final String SRC_PROTOCOL_TYPE_UDP = "udp";
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/httpMsg/InLongHttpMsgHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/httpMsg/InLongHttpMsgHandler.java
new file mode 100644
index 000000000..924760a7e
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/httpMsg/InLongHttpMsgHandler.java
@@ -0,0 +1,385 @@
+/*
+ * 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.httpMsg;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.flume.ChannelException;
+import org.apache.flume.Event;
+import org.apache.flume.event.EventBuilder;
+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.InLongMsg;
+import org.apache.inlong.common.util.NetworkUtils;
+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.source2.BaseSource;
+import org.apache.inlong.dataproxy.utils.AddressUtils;
+import org.apache.inlong.dataproxy.utils.DateTimeUtils;
+import org.apache.inlong.dataproxy.utils.InLongMsgVer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLDecoder;
+import java.util.HashMap;
+import java.util.Map;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.CharsetUtil;
+
+import static io.netty.handler.codec.http.HttpUtil.is100ContinueExpected;
+
+/**
+ * HTTP Server message handler
+ */
+public class InLongHttpMsgHandler extends
SimpleChannelInboundHandler<FullHttpRequest> {
+
+ private static final String hbSrvUrl = "/dataproxy/heartbeat";
+ private static final String msgSrvUrl = "/dataproxy/message";
+
+ private static final Logger logger =
LoggerFactory.getLogger(InLongHttpMsgHandler.class);
+ // log print count
+ private static final LogCounter logCounter = new LogCounter(10, 100000, 30
* 1000);
+
+ private final BaseSource source;
+
+ /**
+ * Constructor
+ *
+ * @param source AbstractSource
+ */
+ public InLongHttpMsgHandler(BaseSource source) {
+ this.source = source;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest
req) throws Exception {
+ // check request decode result
+ if (!req.decoderResult().isSuccess()) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_DECFAIL);
+ sendErrorMsg(ctx, HttpResponseStatus.BAD_REQUEST, "Decode message
failure!");
+ return;
+ }
+ // check request method
+ if (req.method() != HttpMethod.GET && req.method() != HttpMethod.POST)
{
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_INVALIDMETHOD);
+ sendErrorMsg(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED, "Only
support Get and Post methods");
+ return;
+ }
+ // process 100-continue request
+ if (is100ContinueExpected(req)) {
+ ctx.write(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.CONTINUE));
+ }
+ // get requested service
+ String reqUri = req.uri();
+ if (StringUtils.isBlank(reqUri)) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_BLANKURI);
+ sendErrorMsg(ctx, HttpResponseStatus.BAD_REQUEST, "Uri is blank!");
+ return;
+ }
+ try {
+ reqUri = URLDecoder.decode(reqUri, "UTF-8");
+ } catch (UnsupportedEncodingException e) {
+ try {
+ reqUri = URLDecoder.decode(reqUri, "ISO-8859-1");
+ } catch (UnsupportedEncodingException e1) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_URIDECFAIL);
+ sendErrorMsg(ctx, HttpResponseStatus.BAD_REQUEST, "Decode uri
failure!");
+ return;
+ }
+ }
+ // check requested service url
+ if (!reqUri.startsWith(hbSrvUrl) || !reqUri.startsWith(msgSrvUrl)) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_INVALIDURI);
+ sendErrorMsg(ctx, HttpResponseStatus.NOT_IMPLEMENTED, "Not
supported uri!");
+ return;
+ }
+ // get current time and clientIP
+ long msgRcvTime = System.currentTimeMillis();
+ String clientIp = AddressUtils.getChannelRemoteIP(ctx.channel());
+ // check illegal ip
+ if (ConfigManager.getInstance().needChkIllegalIP()
+ && ConfigManager.getInstance().isIllegalIP(clientIp)) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_ILLEGAL_VISIT);
+ sendResponse(ctx, DataProxyErrCode.ILLEGAL_VISIT_IP, true);
+ return;
+ }
+ // check service status.
+ if (source.isRejectService()) {
+ source.fileMetricEventInc(StatConstants.EVENT_SERVICE_CLOSED);
+ sendResponse(ctx, DataProxyErrCode.SERVICE_CLOSED, true);
+ return;
+ }
+ // check sink service status
+ if (!ConfigManager.getInstance().isMqClusterReady()) {
+ source.fileMetricEventInc(StatConstants.EVENT_SERVICE_UNREADY);
+ sendResponse(ctx, DataProxyErrCode.SINK_SERVICE_UNREADY, true);
+ return;
+ }
+ // process hb service
+ if (reqUri.startsWith(hbSrvUrl)) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_HB_SUCCESS);
+ sendResponse(ctx, DataProxyErrCode.SUCCESS, checkClose(req));
+ return;
+ }
+ // process message request
+ processMessage(ctx, req, msgRcvTime, clientIp);
+ }
+
+ private boolean processMessage(ChannelHandlerContext ctx, FullHttpRequest
req,
+ long msgRcvTime, String clientIp) throws Exception {
+ // get and check groupId
+ HttpHeaders headers = req.headers();
+ StringBuilder strBuff = new StringBuilder(512);
+ String groupId = headers.get(AttributeConstants.GROUP_ID);
+ if (StringUtils.isEmpty(groupId)) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_WITHOUTGROUPID);
+ sendResponse(ctx,
DataProxyErrCode.MISS_REQUIRED_GROUPID_ARGUMENT.getErrCode(),
+ strBuff.append("Field
").append(AttributeConstants.GROUP_ID)
+ .append(" must exist and not blank!").toString(),
+ checkClose(req));
+ return false;
+ }
+ // get and check streamId
+ String streamId = headers.get(AttributeConstants.STREAM_ID);
+ if (StringUtils.isEmpty(streamId)) {
+
source.fileMetricEventInc(StatConstants.EVENT_HTTP_WITHOUTSTREAMID);
+ sendResponse(ctx,
DataProxyErrCode.MISS_REQUIRED_STREAMID_ARGUMENT.getErrCode(),
+ strBuff.append("Field
").append(AttributeConstants.STREAM_ID)
+ .append(" must exist and not blank!").toString(),
+ checkClose(req));
+ return false;
+ }
+ // get and check topicName
+ String topicName = ConfigManager.getInstance().getTopicName(groupId,
streamId);
+ if (StringUtils.isBlank(topicName)) {
+ if (CommonConfigHolder.getInstance().isNoTopicAccept()) {
+ source.fileMetricEventInc(StatConstants.EVENT_NOTOPIC);
+ sendResponse(ctx, DataProxyErrCode.TOPIC_IS_BLANK.getErrCode(),
+ strBuff.append("Topic is null for
").append(AttributeConstants.GROUP_ID)
+ .append("(").append(groupId).append("),")
+ .append(AttributeConstants.STREAM_ID)
+
.append("(,").append(streamId).append(")").toString(),
+ checkClose(req));
+ return false;
+ }
+ topicName = source.getDefTopic();
+ }
+ // get and check dt
+ long dataTime = msgRcvTime;
+ String dt = headers.get(AttributeConstants.DATA_TIME);
+ if (StringUtils.isNotEmpty(dt)) {
+ try {
+ dataTime = Long.parseLong(dt);
+ } catch (Throwable e) {
+ //
+ }
+ }
+ // get char set
+ String charset = headers.get(AttrConstants.CHARSET);
+ if (StringUtils.isBlank(charset)) {
+ charset = AttrConstants.CHARSET;
+ }
+ // get and check body
+ String body = headers.get(AttrConstants.BODY);
+ if (StringUtils.isBlank(body)) {
+ if (body == null) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_NOBODY);
+ sendResponse(ctx,
DataProxyErrCode.MISS_REQUIRED_BODY_ARGUMENT.getErrCode(),
+ strBuff.append("Field ").append(AttrConstants.BODY)
+ .append(" is not exist!").toString(),
+ checkClose(req));
+ } else {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_EMPTYBODY);
+ sendResponse(ctx, DataProxyErrCode.EMPTY_MSG.getErrCode(),
+ strBuff.append("Field ").append(AttrConstants.BODY)
+ .append(" is Blank!").toString(),
+ checkClose(req));
+ }
+ return false;
+ }
+ if (body.length() > source.getMaxMsgLength()) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_BODYOVERMAXLEN);
+ sendResponse(ctx,
DataProxyErrCode.BODY_EXCEED_MAX_LEN.getErrCode(),
+ strBuff.append("Error msg, the body
length(").append(body.length())
+ .append(") is bigger than allowed length(")
+
.append(source.getMaxMsgLength()).append(")").toString(),
+ checkClose(req));
+ return false;
+ }
+ // get message count
+ String strMsgCount = headers.get(AttributeConstants.MESSAGE_COUNT);
+ int intMsgCnt = NumberUtils.toInt(strMsgCount, 1);
+ strMsgCount = String.valueOf(intMsgCnt);
+ // build message attributes
+ InLongMsg inLongMsg = InLongMsg.newInLongMsg(source.isCompressed());
+ strBuff.append("&groupId=").append(groupId)
+ .append("&streamId=").append(streamId)
+ .append("&dt=").append(dataTime)
+ .append("&NodeIP=").append(clientIp)
+ .append("&cnt=").append(strMsgCount)
+ .append("&rt=").append(msgRcvTime)
+
.append(AttributeConstants.SEPARATOR).append(AttributeConstants.MSG_RPT_TIME)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(msgRcvTime);
+ inLongMsg.addMsg(strBuff.toString(), body.getBytes(charset));
+ byte[] inlongMsgData = inLongMsg.buildArray();
+ inLongMsg.reset();
+ strBuff.delete(0, strBuff.length());
+ // build flume event
+ Map<String, String> eventHeaders = new HashMap<>();
+ eventHeaders.put(AttributeConstants.GROUP_ID, groupId);
+ eventHeaders.put(AttributeConstants.STREAM_ID, streamId);
+ eventHeaders.put(ConfigConstants.TOPIC_KEY, topicName);
+ eventHeaders.put(AttributeConstants.DATA_TIME,
String.valueOf(dataTime));
+ eventHeaders.put(ConfigConstants.REMOTE_IP_KEY, clientIp);
+ eventHeaders.put(ConfigConstants.MSG_COUNTER_KEY, strMsgCount);
+ eventHeaders.put(ConfigConstants.MSG_ENCODE_VER,
InLongMsgVer.INLONG_V0.getName());
+ eventHeaders.put(AttributeConstants.RCV_TIME,
String.valueOf(msgRcvTime));
+ Event event = EventBuilder.withBody(inlongMsgData, eventHeaders);
+ String msgProcType = "b2b";
+ // build metric data item
+ dataTime = dataTime / 1000 / 60 / 10;
+ dataTime = dataTime * 1000 * 60 * 10;
+
strBuff.append("http").append(AttrConstants.SEP_HASHTAG).append(topicName)
+ .append(AttrConstants.SEP_HASHTAG).append(streamId)
+ .append(AttrConstants.SEP_HASHTAG).append(clientIp)
+
.append(AttrConstants.SEP_HASHTAG).append(NetworkUtils.getLocalIp())
+ .append(AttrConstants.SEP_HASHTAG).append(msgProcType)
+
.append(AttrConstants.SEP_HASHTAG).append(DateTimeUtils.ms2yyyyMMddHHmm(dataTime))
+
.append(AttrConstants.SEP_HASHTAG).append(DateTimeUtils.ms2yyyyMMddHHmm(msgRcvTime));
+ try {
+ source.getChannelProcessor().processEvent(event);
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_POST_SUCCESS);
+ source.fileMetricRecordAdd(strBuff.toString(), intMsgCnt, 1,
body.length(), 0);
+ strBuff.delete(0, strBuff.length());
+ source.addMetric(true, event.getBody().length, event);
+ sendResponse(ctx, DataProxyErrCode.SUCCESS, false);
+ return true;
+ } catch (ChannelException ex) {
+ source.fileMetricEventInc(StatConstants.EVENT_HTTP_POST_DROPPED);
+ source.fileMetricRecordAdd(strBuff.toString(), 0, 0, 0, intMsgCnt);
+ source.addMetric(false, event.getBody().length, event);
+ strBuff.delete(0, strBuff.length());
+ sendResponse(ctx, DataProxyErrCode.UNKNOWN_ERROR.getErrCode(),
+ strBuff.append("Put event to channel failure:
").append(ex.getMessage())
+ .toString(),
+ false);
+ if (logCounter.shouldPrint()) {
+ logger.error("Error write event to channel, data will
discard.", ex);
+ }
+ return false;
+ }
+ }
+
+ @Override
+ public void channelReadComplete(ChannelHandlerContext ctx) {
+ ctx.flush();
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ String clientIp = AddressUtils.getChannelRemoteIP(ctx.channel());
+ logger.error("Http process client={} error, cause:{}, msg:{}",
+ cause, clientIp, cause.getMessage());
+ sendErrorMsg(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR,
+ "Process message failure: " + cause.getMessage());
+ ctx.close();
+ }
+
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
throws Exception {
+ if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
+ ctx.close();
+ }
+ }
+
+ private boolean checkClose(FullHttpRequest req) {
+ String connStatus = req.headers().get("Connection");
+ return !StringUtils.isBlank(connStatus) &&
"close".equalsIgnoreCase(connStatus);
+ }
+
+ private void sendErrorMsg(ChannelHandlerContext ctx, HttpResponseStatus
status, String errMsg) {
+ FullHttpResponse response = new
DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status,
+ Unpooled.copiedBuffer("Failure: " + status + ", "
+ + errMsg + "\r\n", CharsetUtil.UTF_8));
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain;
charset=UTF-8");
+ ctx.writeAndFlush(response).addListener(new SendResultListener(true));
+ }
+
+ private void sendResponse(ChannelHandlerContext ctx, DataProxyErrCode
errCodeObj, boolean isClose) {
+ sendResponse(ctx, errCodeObj.getErrCode(), errCodeObj.getErrMsg(),
isClose);
+
+ }
+
+ private void sendResponse(ChannelHandlerContext ctx, int errCode, String
errMsg, boolean isClose) {
+ FullHttpResponse response = new
DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE,
"application/json;charset=utf-8");
+ StringBuilder builder =
+ new StringBuilder().append("{\"code\":\"").append(errCode)
+ .append("\",\"msg\":\"").append(errMsg).append("\"}");
+ ByteBuf buffer = Unpooled.copiedBuffer(builder.toString(),
CharsetUtil.UTF_8);
+ response.headers().set(HttpHeaderNames.CONTENT_LENGTH,
buffer.readableBytes());
+ response.content().writeBytes(buffer);
+ buffer.release();
+ ctx.writeAndFlush(response).addListener(new
SendResultListener(isClose));
+ }
+
+ private class SendResultListener implements ChannelFutureListener {
+
+ private final boolean isClose;
+
+ public SendResultListener(boolean isClose) {
+ this.isClose = isClose;
+ }
+
+ @Override
+ public void operationComplete(ChannelFuture channelFuture) throws
Exception {
+ if (!channelFuture.isSuccess()) {
+ Throwable throwable = channelFuture.cause();
+ String clientIp =
AddressUtils.getChannelRemoteIP(channelFuture.channel());
+ if (logCounter.shouldPrint()) {
+ logger.error("Http return response to client {} failed,
exception:{}, errmsg:{}",
+ clientIp, throwable,
throwable.getLocalizedMessage());
+ }
+ channelFuture.channel().close();
+ }
+ if (isClose) {
+ channelFuture.channel().close();
+ }
+ }
+ }
+}
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
index 200228ca8..df67b6ac6 100644
---
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
@@ -33,6 +33,7 @@ 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;
@@ -45,7 +46,7 @@ public abstract class AbsV0MsgCodec {
// map joiner
protected static final Joiner.MapJoiner mapJoiner =
Joiner.on(AttributeConstants.SEPARATOR)
.withKeyValueSeparator(AttributeConstants.KEY_VALUE_SEPARATOR);
-
+ protected final Map<String, String> attrMap = new HashMap<>();
protected DataProxyErrCode errCode = DataProxyErrCode.UNKNOWN_ERROR;
protected String errMsg = "";
protected String strRemoteIP;
@@ -63,7 +64,6 @@ public abstract class AbsV0MsgCodec {
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) {
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
index a93a4fbae..7dfef1ec0 100644
---
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
@@ -31,9 +31,11 @@ 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;
@@ -78,7 +80,7 @@ public class CodecBinMsg extends AbsV0MsgCodec {
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.dataTimeMs = this.dataTimeSec * 1000L;
this.msgCount = cb.getUnsignedShort(BIN_MSG_CNT_OFFSET);
this.msgCount = (this.msgCount != 0) ? this.msgCount : 1;
this.uniq = cb.getUnsignedInt(BIN_MSG_UNIQ_OFFSET);
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
index 9d521ccb1..8ad7e0a00 100644
---
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
@@ -29,8 +29,10 @@ 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;
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v1msg/InlongTcpSourceCallback.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v1msg/InlongTcpSourceCallback.java
new file mode 100644
index 000000000..5c77ed328
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v1msg/InlongTcpSourceCallback.java
@@ -0,0 +1,116 @@
+/*
+ * 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.v1msg;
+
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MessagePackHeader;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.ResponseInfo;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.ResultCode;
+import org.apache.inlong.sdk.commons.protocol.SourceCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ * InlongTcpEventCallback
+ */
+public class InlongTcpSourceCallback implements SourceCallback {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(InlongTcpSourceCallback.class);
+
+ private final ChannelHandlerContext ctx;
+ private final MessagePackHeader header;
+ private final CountDownLatch latch;
+ private final AtomicBoolean hasResponsed = new AtomicBoolean(false);
+
+ /**
+ * Constructor
+ *
+ * @param ctx
+ * @param header
+ */
+ public InlongTcpSourceCallback(ChannelHandlerContext ctx,
MessagePackHeader header) {
+ this.ctx = ctx;
+ this.header = header;
+ this.latch = new CountDownLatch(1);
+ }
+
+ /**
+ * callback
+ *
+ * @param resultCode
+ */
+ @Override
+ public void callback(ResultCode resultCode) {
+ // If DataProxy have sent timeout response to SDK, DataProxy do not
send success response to SDK again when
+ // event is success to save.
+ if (this.hasResponsed.getAndSet(true)) {
+ return;
+ }
+ // response
+ try {
+ ResponseInfo.Builder builder = ResponseInfo.newBuilder();
+ builder.setResult(resultCode);
+ 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 {
+ LOG.warn("the send buffer2 is full, so disconnect it!"
+ + "please check remote client; Connection info:{}",
+ remoteChannel);
+ buffer.release();
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ } finally {
+ // notice TCP session
+ this.latch.countDown();
+ }
+ }
+
+ /**
+ * get hasResponsed
+ *
+ * @return the hasResponsed
+ */
+ public AtomicBoolean getHasResponsed() {
+ return hasResponsed;
+ }
+
+ /**
+ * get latch
+ *
+ * @return the latch
+ */
+ public CountDownLatch getLatch() {
+ return latch;
+ }
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/ConfStringUtils.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/ConfStringUtils.java
index ab00a304f..e90cb33d1 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/ConfStringUtils.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/ConfStringUtils.java
@@ -17,10 +17,14 @@
package org.apache.inlong.dataproxy.utils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flume.Context;
+
public class ConfStringUtils {
/**
* isValidIp
+ *
* @param ip
* @return
*/
@@ -54,13 +58,59 @@ public class ConfStringUtils {
/**
* isValidPort
+ *
* @param port
* @return
*/
public static boolean isValidPort(int port) {
- if (port < 0 || port > 65535) {
- return false;
+ return port >= 0 && port <= 65535;
+ }
+
+ /**
+ * 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 static 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 static 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 true;
+ return defVal;
}
}
