You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2021/12/06 02:40:15 UTC
[GitHub] [incubator-inlong] inter12 commented on a change in pull request #1905: [INLONG][feature][audit] add audit-source module
inter12 commented on a change in pull request #1905:
URL: https://github.com/apache/incubator-inlong/pull/1905#discussion_r762667746
##########
File path: inlong-audit/audit-source/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
##########
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.audit.source;
+
+import com.google.common.base.Preconditions;
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.util.concurrent.Executors;
+import org.apache.commons.lang.StringUtils;
+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.audit.base.NamedThreadFactory;
+import org.apache.inlong.audit.consts.ConfigConstants;
+import org.jboss.netty.bootstrap.ServerBootstrap;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFactory;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.group.ChannelGroup;
+import org.jboss.netty.channel.group.DefaultChannelGroup;
+import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
+import org.jboss.netty.util.ThreadNameDeterminer;
+import org.jboss.netty.util.ThreadRenamingRunnable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Simple tcp source
+ *
+ */
+public class SimpleTcpSource extends AbstractSource implements Configurable, EventDrivenSource {
+
+ private static final Logger logger = LoggerFactory.getLogger(SimpleTcpSource.class);
+ private static final String CONNECTIONS = "connections";
+
+ protected int maxConnections = Integer.MAX_VALUE;
+ private ServerBootstrap serverBootstrap = null;
+ protected ChannelGroup allChannels;
+ protected int port;
+ protected String host = null;
+ protected String msgFactoryName;
+ protected String serviceDecoderName;
+ protected String messageHandlerName;
+ protected int maxMsgLength;
+ private int maxThreads = 32;
+
+ private boolean tcpNoDelay = true;
+ private boolean keepAlive = true;
+ private int receiveBufferSize;
+ private int highWaterMark;
+ private int sendBufferSize;
+ private int trafficClass;
+
+ private Channel nettyChannel = null;
+
+ public SimpleTcpSource() {
+ super();
+ allChannels = new DefaultChannelGroup();
+ }
+
+ @Override
+ public synchronized void start() {
+ logger.info("start " + this.getName());
+ super.start();
+
+ ThreadRenamingRunnable.setThreadNameDeterminer(ThreadNameDeterminer.CURRENT);
+ ChannelFactory factory = new NioServerSocketChannelFactory(Executors
+ .newCachedThreadPool(
+ new NamedThreadFactory("tcpSource-nettyBoss-threadGroup")),
+ 1,
+ Executors.newCachedThreadPool(
+ new NamedThreadFactory("tcpSource-nettyWorker-threadGroup")),
+ maxThreads);
+ logger.info("Set max workers : {} ;", maxThreads);
+ ChannelPipelineFactory fac = null;
+
+ serverBootstrap = new ServerBootstrap(factory);
+ serverBootstrap.setOption("child.tcpNoDelay", tcpNoDelay);
+ serverBootstrap.setOption("child.keepAlive", keepAlive);
+ serverBootstrap.setOption("child.receiveBufferSize", receiveBufferSize);
+ serverBootstrap.setOption("child.sendBufferSize", sendBufferSize);
+ serverBootstrap.setOption("child.trafficClass", trafficClass);
+ serverBootstrap.setOption("child.writeBufferHighWaterMark", highWaterMark);
+ logger.info("load msgFactory=" + msgFactoryName + " and serviceDecoderName="
+ + serviceDecoderName);
+ try {
+
+ ServiceDecoder serviceDecoder =
+ (ServiceDecoder) Class.forName(serviceDecoderName).newInstance();
+
+ Class<? extends ChannelPipelineFactory> clazz =
+ (Class<? extends ChannelPipelineFactory>) Class.forName(msgFactoryName);
+
+ Constructor ctor =
+ clazz.getConstructor(AbstractSource.class, ChannelGroup.class,
+ ServiceDecoder.class, String.class,
+ Integer.class, Integer.class, String.class);
+
+ logger.info("Using channel processor:{}", this.getClass().getName());
+ fac = (ChannelPipelineFactory) ctor
+ .newInstance(this, allChannels, serviceDecoder,
+ messageHandlerName, maxMsgLength, maxConnections, this.getName());
+
+ } catch (Exception e) {
+ logger.error(
+ "Simple Tcp Source start error, fail to construct ChannelPipelineFactory with name {}, ex {}",
+ msgFactoryName, e);
+ stop();
+ throw new FlumeException(e.getMessage());
+ }
+
+ serverBootstrap.setPipelineFactory(fac);
+
+ try {
+ if (host == null) {
+ nettyChannel = serverBootstrap.bind(new InetSocketAddress(port));
+ } else {
+ nettyChannel = serverBootstrap.bind(new InetSocketAddress(host, port));
+ }
+ } catch (Exception e) {
+ logger.error("Simple TCP Source error bind host {} port {},program will exit!", host,
+ port);
+ System.exit(-1);
+ }
+
+ allChannels.add(nettyChannel);
+
+ logger.info("Simple TCP Source started at host {}, port {}", host, port);
+
+ }
+
+ @Override
+ public synchronized void stop() {
+ logger.info("[STOP SOURCE]{} stopping...", super.getName());
+ if (allChannels != null && !allChannels.isEmpty()) {
+ try {
+ allChannels.unbind().awaitUninterruptibly();
+ allChannels.close().awaitUninterruptibly();
+ } catch (Exception e) {
+ logger.warn("Simple TCP Source netty server stop ex", e);
+ } finally {
+ allChannels.clear();
+ // allChannels = null;
+ }
+ }
+
+ if (serverBootstrap != null) {
+ try {
+
+ serverBootstrap.releaseExternalResources();
+ } catch (Exception e) {
+ logger.warn("Simple TCP Source serverBootstrap stop ex ", e);
+ } finally {
+ serverBootstrap = null;
+ }
+ }
+
+ super.stop();
+ logger.info("[STOP SOURCE]{} stopped", super.getName());
+ }
+
+ @Override
+ public void configure(Context context) {
+ logger.info("context is {}", context);
+ port = context.getInteger(ConfigConstants.CONFIG_PORT);
+ host = context.getString(ConfigConstants.CONFIG_HOST, "0.0.0.0");
+
+ tcpNoDelay = context.getBoolean(ConfigConstants.TCP_NO_DELAY, true);
+
+ keepAlive = context.getBoolean(ConfigConstants.KEEP_ALIVE, true);
+ highWaterMark = context.getInteger(ConfigConstants.HIGH_WATER_MARK, 64 * 1024);
+ receiveBufferSize = context.getInteger(ConfigConstants.RECEIVE_BUFFER_SIZE, 1024 * 64);
+ if (receiveBufferSize > 16 * 1024 * 1024) {
+ receiveBufferSize = 16 * 1024 * 1024;
+ }
+ Preconditions.checkArgument(receiveBufferSize > 0, "receiveBufferSize must be > 0");
+
+ sendBufferSize = context.getInteger(ConfigConstants.SEND_BUFFER_SIZE, 1024 * 64);
+ if (sendBufferSize > 16 * 1024 * 1024) {
+ sendBufferSize = 16 * 1024 * 1024;
Review comment:
magic number
##########
File path: inlong-audit/audit-source/src/main/java/org/apache/inlong/audit/utils/LogCounter.java
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.audit.utils;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class LogCounter {
+
+ private AtomicInteger counter = new AtomicInteger(0);
+
+ private int start = 10;
+ private int control = 1000;
+ private int reset = 60 * 1000;
+
+ private long lastLogTime = System.currentTimeMillis();
+
+ public LogCounter(int start, int control, int reset) {
+ this.start = start;
+ this.control = control;
+ this.reset = reset;
+ }
+
+ public boolean shouldPrint() {
+ if (System.currentTimeMillis() - lastLogTime > reset) {
Review comment:
recommend to use now.clock instead of System.currentTimeMillis() - lastLogTime > reset) {
##########
File path: inlong-audit/audit-source/src/main/java/org/apache/inlong/audit/sink/pulsar/PulsarClientService.java
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.audit.sink.pulsar;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.FlumeException;
+import org.apache.inlong.audit.consts.AttributeConstants;
+import org.apache.inlong.audit.sink.EventStat;
+import org.apache.inlong.audit.utils.LogCounter;
+import org.apache.inlong.audit.utils.NetworkUtils;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PulsarClientService {
+
+ private static final Logger logger = LoggerFactory.getLogger(PulsarClientService.class);
+
+ private static final LogCounter logPrinterA = new LogCounter(10, 100000, 60 * 1000);
+
+ /*
+ * properties key for pulsar client
+ */
+ private static String PULSAR_SERVER_URL = "pulsar_server_url";
+
+ /*
+ * properties key pulsar producer
+ */
+ private static String SEND_TIMEOUT = "send_timeout_ms";
+ private static String CLIENT_TIMEOUT = "client_op_timeout_second";
+ private static String ENABLE_BATCH = "enable_batch";
+ private static String BLOCK_IF_QUEUE_FULL = "block_if_queue_full";
+ private static String MAX_PENDING_MESSAGES = "max_pending_messages";
+ private static String MAX_BATCHING_MESSAGES = "max_batching_messages";
+
+ private static int DEFAULT_SEND_TIMEOUT_MILL = 30 * 1000;
+ private static int DEFAULT_CLIENT_TIMEOUT_SECOND = 30;
+ private static boolean DEFAULT_ENABLE_BATCH = true;
+ private static boolean DEFAULT_BLOCK_IF_QUEUE_FULL = true;
+ private static int DEFAULT_MAX_PENDING_MESSAGES = 10000;
+ private static int DEFAULT_MAX_BATCHING_MESSAGES = 1000;
+
+ /*
+ * for producer
+ */
+ private Integer sendTimeout; // in millsec
+ private Integer clientOpTimeout;
+ private boolean enableBatch = true;
+ private boolean blockIfQueueFull = true;
+ private int maxPendingMessages = 10000;
+ private int maxBatchingMessages = 1000;
+ public ConcurrentHashMap<String, Producer> producerInfoMap;
+ public PulsarClient pulsarClient;
+ public String pulsarServerUrl;
+
+ private String localIp = "127.0.0.1";
+
+ /**
+ * pulsar client service
+ * @param context
+ */
+ public PulsarClientService(Context context) {
+
+ pulsarServerUrl = context.getString(PULSAR_SERVER_URL);
+ Preconditions.checkState(pulsarServerUrl != null, "No pulsar server url specified");
+
+ sendTimeout = context.getInteger(SEND_TIMEOUT, DEFAULT_SEND_TIMEOUT_MILL);
+ clientOpTimeout = context.getInteger(CLIENT_TIMEOUT, DEFAULT_CLIENT_TIMEOUT_SECOND);
+ logger.debug("PulsarClientService " + SEND_TIMEOUT + " " + sendTimeout);
+ Preconditions.checkArgument(sendTimeout > 0, "sendTimeout must be > 0");
+
+ enableBatch = context.getBoolean(ENABLE_BATCH, DEFAULT_ENABLE_BATCH);
+ blockIfQueueFull = context.getBoolean(BLOCK_IF_QUEUE_FULL, DEFAULT_BLOCK_IF_QUEUE_FULL);
+ maxPendingMessages = context.getInteger(MAX_PENDING_MESSAGES, DEFAULT_MAX_PENDING_MESSAGES);
+ maxBatchingMessages = context.getInteger(MAX_BATCHING_MESSAGES, DEFAULT_MAX_BATCHING_MESSAGES);
+ producerInfoMap = new ConcurrentHashMap<>();
+ localIp = NetworkUtils.getLocalIp();
+
+ }
+
+ /**
+ * init connection
+ * @param callBack
+ */
+ public void initCreateConnection(CreatePulsarClientCallBack callBack) {
+ try {
+ createConnection(callBack);
+ } catch (FlumeException e) {
+ logger.error("Unable to create pulsar client" + ". Exception follows.", e);
+ close();
+ }
+ }
+
+ /**
+ * send message
+ * @param topic
+ * @param event
+ * @param sendMessageCallBack
+ * @param es
+ * @return
+ */
+ public boolean sendMessage(String topic, Event event,
+ SendMessageCallBack sendMessageCallBack, EventStat es) {
+ Producer producer = null;
+ try {
+ producer = getProducer(topic);
+ } catch (Exception e) {
+ if (logPrinterA.shouldPrint()) {
+ logger.error("Get producer failed!", e);
+ }
+ }
+
+ if (producer == null) {
+ logger.error("Get producer is null!");
+ return false;
+ }
+
+ Map<String, String> proMap = new HashMap<>();
+ proMap.put("auditIp", localIp);
+ String streamId = "";
+ String groupId = "";
+ if (event.getHeaders().containsKey(AttributeConstants.INLONG_STREAM_ID)) {
+ streamId = event.getHeaders().get(AttributeConstants.INLONG_STREAM_ID);
+ proMap.put(AttributeConstants.INLONG_STREAM_ID, streamId);
+ }
+ if (event.getHeaders().containsKey(AttributeConstants.INLONG_GROUP_ID)) {
+ groupId = event.getHeaders().get(AttributeConstants.INLONG_GROUP_ID);
+ proMap.put(AttributeConstants.INLONG_GROUP_ID, groupId);
+ }
+
+ logger.debug("producer send msg!");
+ producer.newMessage().properties(proMap).value(event.getBody())
+ .sendAsync().thenAccept((msgId) -> {
+ sendMessageCallBack.handleMessageSendSuccess((MessageIdImpl)msgId, es);
+
+ }).exceptionally((e) -> {
+ sendMessageCallBack.handleMessageSendException(es, e);
+ return null;
+ });
+ return true;
+ }
+
+ /**
+ * If this function is called successively without calling {@see #destroyConnection()}, only the
+ * first call has any effect.
+ *
+ * @throws FlumeException if an RPC client connection could not be opened
+ */
+ private void createConnection(CreatePulsarClientCallBack callBack) throws FlumeException {
+ if (pulsarClient != null) {
+ return;
+ }
+ try {
+ pulsarClient = initPulsarClient(pulsarServerUrl);
+ callBack.handleCreateClientSuccess(pulsarServerUrl);
+ } catch (PulsarClientException e) {
+ callBack.handleCreateClientException(pulsarServerUrl);
+ logger.error("create connnection error in metasink, "
+ + "maybe pulsar master set error, please re-check.url{}, ex1 {}",
+ pulsarServerUrl,
+ e.getMessage());
+ } catch (Throwable e) {
+ callBack.handleCreateClientException(pulsarServerUrl);
+ logger.error("create connnection error in metasink, "
+ + "maybe pulsar master set error/shutdown in progress, please "
+ + "re-check. url{}, ex2 {}",
+ pulsarServerUrl,
+ e.getMessage());
+ }
+ }
+
+ private PulsarClient initPulsarClient(String pulsarUrl) throws Exception {
+ return PulsarClient.builder()
+ .serviceUrl(pulsarUrl)
+ .connectionTimeout(clientOpTimeout, TimeUnit.SECONDS)
+ .build();
+ }
+
+ public Producer initTopicProducer(String topic) {
+ logger.info("initTopicProducer topic = {}", topic);
+ Producer producer = null;
+ try {
+ producer = pulsarClient.newProducer().sendTimeout(sendTimeout,
+ TimeUnit.MILLISECONDS)
+ .topic(topic)
+ .enableBatching(enableBatch)
+ .blockIfQueueFull(blockIfQueueFull)
+ .maxPendingMessages(maxPendingMessages)
+ .batchingMaxMessages(maxBatchingMessages)
+ .create();
+ } catch (PulsarClientException e) {
+ logger.error("create pulsar client has error e = {}", e);
Review comment:
init Producer failed，throw exception is better.
##########
File path: inlong-audit/audit-source/src/main/java/org/apache/inlong/audit/base/HighPriorityThreadFactory.java
##########
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.audit.base;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class HighPriorityThreadFactory
+ implements ThreadFactory {
+ static final AtomicInteger poolNumber = new AtomicInteger(1);
+ final AtomicInteger threadNumber;
Review comment:
should add access modifier ，eg. private
##########
File path: inlong-audit/audit-source/src/main/java/org/apache/inlong/audit/sink/pulsar/CreatePulsarClientCallBack.java
##########
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.audit.sink.pulsar;
+
+public interface CreatePulsarClientCallBack {
+
+ void handleCreateClientSuccess(String url);
Review comment:
Missing comment
##########
File path: inlong-audit/audit-source/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
##########
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.audit.source;
+
+import com.google.common.base.Preconditions;
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.util.concurrent.Executors;
+import org.apache.commons.lang.StringUtils;
+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.audit.base.NamedThreadFactory;
+import org.apache.inlong.audit.consts.ConfigConstants;
+import org.jboss.netty.bootstrap.ServerBootstrap;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFactory;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.group.ChannelGroup;
+import org.jboss.netty.channel.group.DefaultChannelGroup;
+import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
+import org.jboss.netty.util.ThreadNameDeterminer;
+import org.jboss.netty.util.ThreadRenamingRunnable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Simple tcp source
+ *
+ */
+public class SimpleTcpSource extends AbstractSource implements Configurable, EventDrivenSource {
+
+ private static final Logger logger = LoggerFactory.getLogger(SimpleTcpSource.class);
+ private static final String CONNECTIONS = "connections";
+
+ protected int maxConnections = Integer.MAX_VALUE;
+ private ServerBootstrap serverBootstrap = null;
+ protected ChannelGroup allChannels;
+ protected int port;
+ protected String host = null;
+ protected String msgFactoryName;
+ protected String serviceDecoderName;
+ protected String messageHandlerName;
+ protected int maxMsgLength;
+ private int maxThreads = 32;
+
+ private boolean tcpNoDelay = true;
+ private boolean keepAlive = true;
+ private int receiveBufferSize;
+ private int highWaterMark;
+ private int sendBufferSize;
+ private int trafficClass;
+
+ private Channel nettyChannel = null;
+
+ public SimpleTcpSource() {
+ super();
+ allChannels = new DefaultChannelGroup();
+ }
+
+ @Override
+ public synchronized void start() {
+ logger.info("start " + this.getName());
+ super.start();
+
+ ThreadRenamingRunnable.setThreadNameDeterminer(ThreadNameDeterminer.CURRENT);
+ ChannelFactory factory = new NioServerSocketChannelFactory(Executors
+ .newCachedThreadPool(
+ new NamedThreadFactory("tcpSource-nettyBoss-threadGroup")),
+ 1,
+ Executors.newCachedThreadPool(
+ new NamedThreadFactory("tcpSource-nettyWorker-threadGroup")),
+ maxThreads);
+ logger.info("Set max workers : {} ;", maxThreads);
+ ChannelPipelineFactory fac = null;
+
+ serverBootstrap = new ServerBootstrap(factory);
+ serverBootstrap.setOption("child.tcpNoDelay", tcpNoDelay);
+ serverBootstrap.setOption("child.keepAlive", keepAlive);
+ serverBootstrap.setOption("child.receiveBufferSize", receiveBufferSize);
+ serverBootstrap.setOption("child.sendBufferSize", sendBufferSize);
+ serverBootstrap.setOption("child.trafficClass", trafficClass);
+ serverBootstrap.setOption("child.writeBufferHighWaterMark", highWaterMark);
+ logger.info("load msgFactory=" + msgFactoryName + " and serviceDecoderName="
+ + serviceDecoderName);
+ try {
+
+ ServiceDecoder serviceDecoder =
+ (ServiceDecoder) Class.forName(serviceDecoderName).newInstance();
+
+ Class<? extends ChannelPipelineFactory> clazz =
+ (Class<? extends ChannelPipelineFactory>) Class.forName(msgFactoryName);
+
+ Constructor ctor =
+ clazz.getConstructor(AbstractSource.class, ChannelGroup.class,
+ ServiceDecoder.class, String.class,
+ Integer.class, Integer.class, String.class);
+
+ logger.info("Using channel processor:{}", this.getClass().getName());
+ fac = (ChannelPipelineFactory) ctor
+ .newInstance(this, allChannels, serviceDecoder,
+ messageHandlerName, maxMsgLength, maxConnections, this.getName());
+
+ } catch (Exception e) {
+ logger.error(
+ "Simple Tcp Source start error, fail to construct ChannelPipelineFactory with name {}, ex {}",
+ msgFactoryName, e);
+ stop();
+ throw new FlumeException(e.getMessage());
+ }
+
+ serverBootstrap.setPipelineFactory(fac);
+
+ try {
+ if (host == null) {
+ nettyChannel = serverBootstrap.bind(new InetSocketAddress(port));
+ } else {
+ nettyChannel = serverBootstrap.bind(new InetSocketAddress(host, port));
+ }
+ } catch (Exception e) {
+ logger.error("Simple TCP Source error bind host {} port {},program will exit!", host,
+ port);
+ System.exit(-1);
+ }
+
+ allChannels.add(nettyChannel);
+
+ logger.info("Simple TCP Source started at host {}, port {}", host, port);
+
+ }
+
+ @Override
+ public synchronized void stop() {
+ logger.info("[STOP SOURCE]{} stopping...", super.getName());
+ if (allChannels != null && !allChannels.isEmpty()) {
+ try {
+ allChannels.unbind().awaitUninterruptibly();
+ allChannels.close().awaitUninterruptibly();
+ } catch (Exception e) {
+ logger.warn("Simple TCP Source netty server stop ex", e);
+ } finally {
+ allChannels.clear();
+ // allChannels = null;
+ }
+ }
+
+ if (serverBootstrap != null) {
+ try {
+
+ serverBootstrap.releaseExternalResources();
+ } catch (Exception e) {
+ logger.warn("Simple TCP Source serverBootstrap stop ex ", e);
+ } finally {
+ serverBootstrap = null;
+ }
+ }
+
+ super.stop();
+ logger.info("[STOP SOURCE]{} stopped", super.getName());
+ }
+
+ @Override
+ public void configure(Context context) {
+ logger.info("context is {}", context);
+ port = context.getInteger(ConfigConstants.CONFIG_PORT);
+ host = context.getString(ConfigConstants.CONFIG_HOST, "0.0.0.0");
+
+ tcpNoDelay = context.getBoolean(ConfigConstants.TCP_NO_DELAY, true);
+
+ keepAlive = context.getBoolean(ConfigConstants.KEEP_ALIVE, true);
+ highWaterMark = context.getInteger(ConfigConstants.HIGH_WATER_MARK, 64 * 1024);
+ receiveBufferSize = context.getInteger(ConfigConstants.RECEIVE_BUFFER_SIZE, 1024 * 64);
+ if (receiveBufferSize > 16 * 1024 * 1024) {
+ receiveBufferSize = 16 * 1024 * 1024;
+ }
+ Preconditions.checkArgument(receiveBufferSize > 0, "receiveBufferSize must be > 0");
+
+ sendBufferSize = context.getInteger(ConfigConstants.SEND_BUFFER_SIZE, 1024 * 64);
+ if (sendBufferSize > 16 * 1024 * 1024) {
+ sendBufferSize = 16 * 1024 * 1024;
+ }
+ Preconditions.checkArgument(sendBufferSize > 0, "sendBufferSize must be > 0");
+
+ trafficClass = context.getInteger(ConfigConstants.TRAFFIC_CLASS, 0);
+ Preconditions.checkArgument((trafficClass == 0 || trafficClass == 96),
Review comment:
magic number
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org