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 dae7074d9e [INLONG-11564][SDK] DataProxy SDK Implementation
Optimization (#11581)
dae7074d9e is described below
commit dae7074d9ee219135138e8a9a8a3db8948366bfd
Author: Goson Zhang <[email protected]>
AuthorDate: Fri Dec 6 08:57:13 2024 +0800
[INLONG-11564][SDK] DataProxy SDK Implementation Optimization (#11581)
Co-authored-by: gosonzhang <[email protected]>
---
.../inlong/sdk/dataproxy/DefaultMessageSender.java | 2 -
.../inlong/sdk/dataproxy/ProxyClientConfig.java | 24 --
.../sdk/dataproxy/codec/ProtocolDecoder.java | 53 ++--
.../sdk/dataproxy/codec/ProtocolEncoder.java | 32 +--
.../inlong/sdk/dataproxy/common/SendResult.java | 2 +-
.../sdk/dataproxy/config/EncryptConfigEntry.java | 5 +-
.../inlong/sdk/dataproxy/config/HostInfo.java | 4 +-
.../sdk/dataproxy/config/ProxyConfigManager.java | 270 +++++++-------------
.../sdk/dataproxy/network/SyncMessageCallable.java | 13 +-
.../sdk/dataproxy/network/TimeScanObject.java | 4 +-
.../dataproxy/threads/ManagerFetcherThread.java | 68 -----
.../inlong/sdk/dataproxy/utils/EventLoopUtil.java | 3 +-
.../sdk/dataproxy/utils/ServiceDiscoveryUtils.java | 279 ---------------------
13 files changed, 153 insertions(+), 606 deletions(-)
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
index 6fa6f6ff9a..dd22e63cce 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
@@ -29,7 +29,6 @@ import
org.apache.inlong.sdk.dataproxy.network.ProxysdkException;
import org.apache.inlong.sdk.dataproxy.network.Sender;
import org.apache.inlong.sdk.dataproxy.network.SequentialID;
import org.apache.inlong.sdk.dataproxy.threads.IndexCollectThread;
-import org.apache.inlong.sdk.dataproxy.threads.ManagerFetcherThread;
import org.apache.inlong.sdk.dataproxy.utils.ProxyUtils;
import org.slf4j.Logger;
@@ -52,7 +51,6 @@ public class DefaultMessageSender implements MessageSender {
private static final ConcurrentHashMap<Integer, DefaultMessageSender>
CACHE_SENDER =
new ConcurrentHashMap<>();
private static final AtomicBoolean MANAGER_FETCHER_THREAD_STARTED = new
AtomicBoolean(false);
- private static ManagerFetcherThread managerFetcherThread;
private static final SequentialID idGenerator = new SequentialID();
private final Sender sender;
private final IndexCollectThread indexCol;
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/ProxyClientConfig.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/ProxyClientConfig.java
index e7a0f6a3e6..ce8a3b3b39 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/ProxyClientConfig.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/ProxyClientConfig.java
@@ -34,7 +34,6 @@ public class ProxyClientConfig {
private String managerIP = "";
private String managerAddress;
- private String managerIpLocalPath = System.getProperty("user.dir") +
"/.inlong/.managerIps";
private String managerUrl = "";
private int proxyUpdateIntervalMinutes;
private int proxyUpdateMaxRetry;
@@ -54,7 +53,6 @@ public class ProxyClientConfig {
private String authSecretKey;
private String protocolType;
- private boolean enableSaveManagerVIps = false;
// metric configure
private MetricConfig metricConfig = new MetricConfig();
@@ -211,28 +209,6 @@ public class ProxyClientConfig {
return managerIP;
}
- public String getManagerIpLocalPath() {
- return managerIpLocalPath;
- }
-
- public void setManagerIpLocalPath(String managerIpLocalPath) throws
ProxysdkException {
- if (StringUtils.isEmpty(managerIpLocalPath)) {
- throw new ProxysdkException("managerIpLocalPath is empty.");
- }
- if (managerIpLocalPath.charAt(managerIpLocalPath.length() - 1) == '/')
{
- managerIpLocalPath = managerIpLocalPath.substring(0,
managerIpLocalPath.length() - 1);
- }
- this.managerIpLocalPath = managerIpLocalPath + "/.managerIps";
- }
-
- public boolean isEnableSaveManagerVIps() {
- return enableSaveManagerVIps;
- }
-
- public void setEnableSaveManagerVIps(boolean enable) {
- this.enableSaveManagerVIps = enable;
- }
-
public String getConfStoreBasePath() {
return confStoreBasePath;
}
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/codec/ProtocolDecoder.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/codec/ProtocolDecoder.java
index 2038a8b8d6..b7c31cba79 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/codec/ProtocolDecoder.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/codec/ProtocolDecoder.java
@@ -17,6 +17,8 @@
package org.apache.inlong.sdk.dataproxy.codec;
+import org.apache.inlong.sdk.dataproxy.utils.LogCounter;
+
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
@@ -29,18 +31,19 @@ import java.util.List;
public class ProtocolDecoder extends MessageToMessageDecoder<ByteBuf> {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ProtocolDecoder.class);
-
+ private static final Logger logger =
LoggerFactory.getLogger(ProtocolDecoder.class);
+ private static final LogCounter decExptCounter = new LogCounter(10,
200000, 60 * 1000L);
@Override
protected void decode(ChannelHandlerContext ctx,
ByteBuf buffer, List<Object> out) throws Exception {
buffer.markReaderIndex();
// totallen
int totalLen = buffer.readInt();
- LOGGER.debug("decode totalLen : {}", totalLen);
if (totalLen != buffer.readableBytes()) {
- LOGGER.error("totalLen is not equal readableBytes.total:" +
totalLen
- + ";readableBytes:" + buffer.readableBytes());
+ if (decExptCounter.shouldPrint()) {
+ logger.error("Length not equal,
totalLen={},readableBytes={},from={}",
+ totalLen, buffer.readableBytes(), ctx.channel());
+ }
buffer.resetReaderIndex();
throw new Exception("totalLen is not equal readableBytes.total");
}
@@ -48,14 +51,17 @@ public class ProtocolDecoder extends
MessageToMessageDecoder<ByteBuf> {
int msgType = buffer.readByte() & 0x1f;
if (msgType == 4) {
- LOGGER.info("debug decode");
- }
- if (msgType == 3 | msgType == 5) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("debug decode");
+ }
+ } else if (msgType == 3 | msgType == 5) {
// bodylen
int bodyLength = buffer.readInt();
if (bodyLength >= totalLen) {
- LOGGER.error("bodyLen is greater than totalLen.totalLen:" +
totalLen
- + ";bodyLen:" + bodyLength);
+ if (decExptCounter.shouldPrint()) {
+ logger.error("bodyLen greater than totalLen,
totalLen={},bodyLen={},from={}",
+ totalLen, bodyLength, ctx.channel());
+ }
buffer.resetReaderIndex();
throw new Exception("bodyLen is greater than
totalLen.totalLen");
}
@@ -64,20 +70,19 @@ public class ProtocolDecoder extends
MessageToMessageDecoder<ByteBuf> {
bodyBytes = new byte[bodyLength];
buffer.readBytes(bodyBytes);
}
-
// attrlen
+ String attrInfo = "";
int attrLength = buffer.readInt();
- byte[] attrBytes = null;
if (attrLength > 0) {
- attrBytes = new byte[attrLength];
+ byte[] attrBytes = new byte[attrLength];
buffer.readBytes(attrBytes);
+ attrInfo = new String(attrBytes, StandardCharsets.UTF_8);
}
EncodeObject object;
if (bodyBytes == null) {
- object = new EncodeObject(new String(attrBytes,
StandardCharsets.UTF_8));
+ object = new EncodeObject(attrInfo);
} else {
- object = new EncodeObject(Collections.singletonList(bodyBytes),
- new String(attrBytes, StandardCharsets.UTF_8));
+ object = new
EncodeObject(Collections.singletonList(bodyBytes), attrInfo);
}
object.setMsgtype(5);
out.add(object);
@@ -85,12 +90,13 @@ public class ProtocolDecoder extends
MessageToMessageDecoder<ByteBuf> {
int seqId = buffer.readInt();
int attrLen = buffer.readShort();
- byte[] attrBytes = null;
+ String attrInfo = "";
if (attrLen > 0) {
- attrBytes = new byte[attrLen];
+ byte[] attrBytes = new byte[attrLen];
buffer.readBytes(attrBytes);
+ attrInfo = new String(attrBytes, StandardCharsets.UTF_8);
}
- EncodeObject object = new EncodeObject(new String(attrBytes,
StandardCharsets.UTF_8));
+ EncodeObject object = new EncodeObject(attrInfo);
object.setMessageId(String.valueOf(seqId));
buffer.readShort();
@@ -103,15 +109,14 @@ public class ProtocolDecoder extends
MessageToMessageDecoder<ByteBuf> {
buffer.skipBytes(4 + 1 + 4); // skip datatime, body_ver and
body_len
final short load = buffer.readShort(); // read from body
int attrLen = buffer.readShort();
- byte[] attrBytes = null;
+ String attrInfo = "";
if (attrLen > 0) {
- attrBytes = new byte[attrLen];
+ byte[] attrBytes = new byte[attrLen];
buffer.readBytes(attrBytes);
+ attrInfo = new String(attrBytes, StandardCharsets.UTF_8);
}
buffer.skipBytes(2); // skip magic
-
- String attrs = (attrBytes == null ? "" : new String(attrBytes,
StandardCharsets.UTF_8));
- EncodeObject object = new EncodeObject(attrs);
+ EncodeObject object = new EncodeObject(attrInfo);
object.setMsgtype(8);
object.setLoad(load);
out.add(object);
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/codec/ProtocolEncoder.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/codec/ProtocolEncoder.java
index c3d463987b..1a20766e55 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/codec/ProtocolEncoder.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/codec/ProtocolEncoder.java
@@ -99,16 +99,17 @@ public class ProtocolEncoder extends
MessageToMessageEncoder<EncodeObject> {
if (object.isAuth()) {
msgType |= FLAG_ALLOW_AUTH;
}
- int totalLength = 1 + 4 + 1 + 4 + 2 +
endAttr.getBytes("utf8").length + 2;
+ byte[] attrData = endAttr.getBytes(StandardCharsets.UTF_8);
+ int totalLength = 1 + 4 + 1 + 4 + 2 + attrData.length + 2;
buf = ByteBufAllocator.DEFAULT.buffer(4 + totalLength);
buf.writeInt(totalLength);
buf.writeByte(msgType);
buf.writeInt((int) object.getDt());
buf.writeByte(1);
buf.writeInt(0);
- buf.writeShort(endAttr.getBytes("utf8").length);
- if (endAttr.getBytes("utf8").length > 0) {
- buf.writeBytes(endAttr.getBytes("utf8"));
+ buf.writeShort(attrData.length);
+ if (attrData.length > 0) {
+ buf.writeBytes(attrData);
}
buf.writeShort(0xee01);
} catch (Throwable ex) {
@@ -160,7 +161,8 @@ public class ProtocolEncoder extends
MessageToMessageEncoder<EncodeObject> {
if (object.isCompress()) {
msgType |= FLAG_ALLOW_COMPRESS;
}
- totalLength = totalLength + body.length +
endAttr.getBytes("utf8").length;
+ byte[] attrData = endAttr.getBytes(StandardCharsets.UTF_8);
+ totalLength = totalLength + body.length + attrData.length;
buf = ByteBufAllocator.DEFAULT.buffer(4 + totalLength);
buf.writeInt(totalLength);
buf.writeByte(msgType);
@@ -181,8 +183,8 @@ public class ProtocolEncoder extends
MessageToMessageEncoder<EncodeObject> {
buf.writeInt(body.length);
buf.writeBytes(body);
- buf.writeShort(endAttr.getBytes("utf8").length);
- buf.writeBytes(endAttr.getBytes("utf8"));
+ buf.writeShort(attrData.length);
+ buf.writeBytes(attrData);
buf.writeShort(0xee01);
}
return buf;
@@ -207,7 +209,7 @@ public class ProtocolEncoder extends
MessageToMessageEncoder<EncodeObject> {
ByteArrayOutputStream data = new ByteArrayOutputStream();
for (byte[] entry : object.getBodylist()) {
if (totalCnt++ > 0) {
- data.write("\n".getBytes("utf8"));
+
data.write(AttributeConstants.LINE_FEED_SEP.getBytes(StandardCharsets.UTF_8));
}
data.write(entry);
}
@@ -280,14 +282,15 @@ public class ProtocolEncoder extends
MessageToMessageEncoder<EncodeObject> {
if (object.isEncrypt()) {
msgType |= FLAG_ALLOW_ENCRYPT;
}
- totalLength = totalLength + body.length +
msgAttrs.getBytes("utf8").length;
+ byte[] attrData = msgAttrs.getBytes(StandardCharsets.UTF_8);
+ totalLength = totalLength + body.length + attrData.length;
buf = ByteBufAllocator.DEFAULT.buffer(4 + totalLength);
buf.writeInt(totalLength);
buf.writeByte(msgType);
buf.writeInt(body.length);
buf.writeBytes(body);
- buf.writeInt(msgAttrs.getBytes("utf8").length);
- buf.writeBytes(msgAttrs.getBytes("utf8"));
+ buf.writeInt(attrData.length);
+ buf.writeBytes(attrData);
}
} catch (Throwable ex) {
if (exptCounter.shouldPrint()) {
@@ -344,14 +347,15 @@ public class ProtocolEncoder extends
MessageToMessageEncoder<EncodeObject> {
if (object.isEncrypt()) {
msgType |= FLAG_ALLOW_ENCRYPT;
}
- totalLength = totalLength + body.length +
msgAttrs.getBytes("utf8").length;
+ byte[] attrData = msgAttrs.getBytes(StandardCharsets.UTF_8);
+ totalLength = totalLength + body.length + attrData.length;
buf = ByteBufAllocator.DEFAULT.buffer(4 + totalLength);
buf.writeInt(totalLength);
buf.writeByte(msgType);
buf.writeInt(body.length);
buf.writeBytes(body);
- buf.writeInt(msgAttrs.getBytes("utf8").length);
- buf.writeBytes(msgAttrs.getBytes("utf8"));
+ buf.writeInt(attrData.length);
+ buf.writeBytes(attrData);
}
} catch (Throwable ex) {
if (exptCounter.shouldPrint()) {
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/common/SendResult.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/common/SendResult.java
index adab601e0a..f336702ee4 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/common/SendResult.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/common/SendResult.java
@@ -18,8 +18,8 @@
package org.apache.inlong.sdk.dataproxy.common;
public enum SendResult {
- INVALID_ATTRIBUTES, // including DataProxyErrCode(100,101,102,112)
OK,
+ INVALID_ATTRIBUTES, // including DataProxyErrCode(100,101,102,112)
TIMEOUT,
CONNECTION_BREAK,
THREAD_INTERRUPT,
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/EncryptConfigEntry.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/EncryptConfigEntry.java
index 6acfe09d8a..47f6cd1ed7 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/EncryptConfigEntry.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/EncryptConfigEntry.java
@@ -25,6 +25,7 @@ import org.slf4j.LoggerFactory;
import java.net.URLEncoder;
import java.security.interfaces.RSAPublicKey;
+import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;
public class EncryptConfigEntry implements java.io.Serializable {
@@ -122,7 +123,7 @@ public class EncryptConfigEntry implements
java.io.Serializable {
@Override
public boolean equals(Object other) {
- if (other == null || !(other instanceof EncryptConfigEntry)) {
+ if (!(other instanceof EncryptConfigEntry)) {
return false;
}
if (other == this) {
@@ -131,7 +132,7 @@ public class EncryptConfigEntry implements
java.io.Serializable {
EncryptConfigEntry info = (EncryptConfigEntry) other;
return (this.userName.equals(info.getUserName()))
&& (this.version.equals(info.getVersion()))
- && (this.pubKey == info.getPubKey());
+ && (Objects.equals(this.pubKey, info.getPubKey()));
}
public String toString() {
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/HostInfo.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/HostInfo.java
index c93c872ea5..071cf4b6eb 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/HostInfo.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/HostInfo.java
@@ -25,10 +25,10 @@ public class HostInfo implements Comparable<HostInfo>,
java.io.Serializable {
private final String hostName;
private final int portNumber;
- public HostInfo(String referenceName, String hostName, int portNumber) {
- this.referenceName = referenceName;
+ public HostInfo(String hostName, int portNumber) {
this.hostName = hostName;
this.portNumber = portNumber;
+ this.referenceName = hostName + ":" + portNumber;
}
public String getReferenceName() {
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/ProxyConfigManager.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/ProxyConfigManager.java
index b10e533037..d259cdff08 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/ProxyConfigManager.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/config/ProxyConfigManager.java
@@ -35,7 +35,6 @@ import com.google.gson.stream.JsonReader;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
-import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.Header;
@@ -91,11 +90,10 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
public class ProxyConfigManager extends Thread {
public static final String APPLICATION_JSON = "application/json";
- private static final Logger LOGGER =
LoggerFactory.getLogger(ProxyConfigManager.class);
+ private static final Logger logger =
LoggerFactory.getLogger(ProxyConfigManager.class);
private final ProxyClientConfig clientConfig;
private final ClientMgr clientManager;
private final ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
- private final JsonParser jsonParser = new JsonParser();
private final Gson gson = new Gson();
private final HashRing hashRing = HashRing.getInstance();
private List<HostInfo> proxyInfoList = new ArrayList<HostInfo>();
@@ -104,8 +102,8 @@ public class ProxyConfigManager extends Thread {
private String inlongGroupId;
private String localMd5;
private boolean bShutDown = false;
- private long doworkTime = 0;
- private EncryptConfigEntry userEncryConfigEntry;
+ private long lstUpdatedTime = 0;
+ private EncryptConfigEntry userEncryptConfigEntry;
public ProxyConfigManager(final ProxyClientConfig configure, final
ClientMgr clientManager) {
this.clientConfig = configure;
@@ -122,7 +120,7 @@ public class ProxyConfigManager extends Thread {
}
public void shutDown() {
- LOGGER.info("Begin to shut down ProxyConfigManager!");
+ logger.info("Begin to shut down ProxyConfigManager!");
bShutDown = true;
}
@@ -132,9 +130,9 @@ public class ProxyConfigManager extends Thread {
try {
doProxyEntryQueryWork();
updateEncryptConfigEntry();
- LOGGER.info("ProxyConf update!");
+ logger.info("ProxyConf update!");
} catch (Throwable e) {
- LOGGER.error("Refresh proxy ip list runs into exception {},
{}", e.toString(), e.getStackTrace());
+ logger.error("Refresh proxy ip list runs into exception {},
{}", e.toString(), e.getStackTrace());
e.printStackTrace();
}
@@ -147,13 +145,13 @@ public class ProxyConfigManager extends Thread {
if (proxyUpdateIntervalSec > 5) {
sleepTimeSec = proxyUpdateIntervalSec + random.nextInt() %
(proxyUpdateIntervalSec / 5);
}
- LOGGER.info("sleep time {}", sleepTimeSec);
+ logger.info("sleep time {}", sleepTimeSec);
Thread.sleep(sleepTimeSec * 1000);
} catch (Throwable e2) {
//
}
}
- LOGGER.info("ProxyConfigManager worker existed!");
+ logger.info("ProxyConfigManager worker existed!");
}
/**
@@ -170,11 +168,11 @@ public class ProxyConfigManager extends Thread {
if (diffTime < clientConfig.getMaxProxyCacheTimeInMs()) {
JsonReader reader = new JsonReader(new
FileReader(configCachePath));
ProxyConfigEntry proxyConfigEntry = gson.fromJson(reader,
ProxyConfigEntry.class);
- LOGGER.info("{} has a backup! {}", inlongGroupId,
proxyConfigEntry);
+ logger.info("{} has a backup! {}", inlongGroupId,
proxyConfigEntry);
return proxyConfigEntry;
}
} catch (Exception ex) {
- LOGGER.warn("try to read local cache, caught {}", ex.getMessage());
+ logger.warn("try to read local cache, caught {}", ex.getMessage());
} finally {
rw.readLock().unlock();
}
@@ -189,13 +187,13 @@ public class ProxyConfigManager extends Thread {
// try to create parent
file.getParentFile().mkdirs();
}
- LOGGER.info("try to write {}} to local cache {}", entry,
configCachePath);
+ logger.info("try to write {}} to local cache {}", entry,
configCachePath);
FileWriter fileWriter = new FileWriter(configCachePath);
gson.toJson(entry, fileWriter);
fileWriter.flush();
fileWriter.close();
} catch (Exception ex) {
- LOGGER.warn("try to write local cache, caught {}",
ex.getMessage());
+ logger.warn("try to write local cache, caught {}",
ex.getMessage());
} finally {
rw.writeLock().unlock();
}
@@ -205,7 +203,7 @@ public class ProxyConfigManager extends Thread {
try {
return requestProxyList(this.clientConfig.getManagerUrl());
} catch (Exception e) {
- LOGGER.warn("try to request proxy list by http, caught {}",
e.getMessage());
+ logger.warn("try to request proxy list by http, caught {}",
e.getMessage());
}
return null;
}
@@ -280,7 +278,7 @@ public class ProxyConfigManager extends Thread {
}
/* We should exit if no local IP list and can't request it from
manager. */
if (localMd5 == null && proxyEntry == null) {
- LOGGER.error("Can't connect manager at the start of proxy API
{}",
+ logger.error("Can't connect manager at the start of proxy API
{}",
this.clientConfig.getManagerUrl());
proxyEntry = tryToReadCacheProxyEntry(configAddr);
}
@@ -290,7 +288,7 @@ public class ProxyConfigManager extends Thread {
s.append(tmp.getHostName()).append(";").append(tmp.getPortNumber())
.append(",");
}
- LOGGER.warn("Backup proxyEntry [{}]", s);
+ logger.warn("Backup proxyEntry [{}]", s);
}
}
if (localMd5 == null && proxyEntry == null && proxyInfoList == null) {
@@ -312,7 +310,7 @@ public class ProxyConfigManager extends Thread {
*/
private void compareProxyList(ProxyConfigEntry proxyEntry) {
if (proxyEntry != null) {
- LOGGER.info("{}", proxyEntry.toString());
+ logger.info("{}", proxyEntry.toString());
if (proxyEntry.getSize() != 0) {
/* Initialize the current proxy information list first. */
clientManager.setLoadThreshold(proxyEntry.getLoad());
@@ -326,31 +324,31 @@ public class ProxyConfigManager extends Thread {
String oldMd5 = calcHostInfoMd5(proxyInfoList);
if (newMd5 != null && !newMd5.equals(oldMd5)) {
/* Choose random alive connections to send messages. */
- LOGGER.info("old md5 {} new md5 {}", oldMd5, newMd5);
+ logger.info("old md5 {} new md5 {}", oldMd5, newMd5);
proxyInfoList.clear();
proxyInfoList = newProxyInfoList;
clientManager.setProxyInfoList(proxyInfoList);
- doworkTime = System.currentTimeMillis();
+ lstUpdatedTime = System.currentTimeMillis();
} else if (proxyEntry.getSwitchStat() != oldStat) {
/* judge cluster's switch state */
oldStat = proxyEntry.getSwitchStat();
- if ((System.currentTimeMillis() - doworkTime) > 3 * 60 *
1000) {
- LOGGER.info("switch the cluster!");
+ if ((System.currentTimeMillis() - lstUpdatedTime) > 3 * 60
* 1000) {
+ logger.info("switch the cluster!");
proxyInfoList.clear();
proxyInfoList = newProxyInfoList;
clientManager.setProxyInfoList(proxyInfoList);
} else {
- LOGGER.info("only change oldStat ");
+ logger.info("only change oldStat ");
}
} else {
newProxyInfoList.clear();
- LOGGER.info("proxy IP list doesn't change, load {}",
proxyEntry.getLoad());
+ logger.info("proxy IP list doesn't change, load {}",
proxyEntry.getLoad());
}
if (clientConfig.getLoadBalance() ==
LoadBalance.CONSISTENCY_HASH) {
updateHashRing(proxyInfoList);
}
} else {
- LOGGER.error("proxyEntry's size is zero");
+ logger.error("proxyEntry's size is zero");
}
}
}
@@ -359,7 +357,7 @@ public class ProxyConfigManager extends Thread {
if (StringUtils.isBlank(userName)) {
return null;
}
- EncryptConfigEntry encryptEntry = this.userEncryConfigEntry;
+ EncryptConfigEntry encryptEntry = this.userEncryptConfigEntry;
if (encryptEntry == null) {
int retryCount = 0;
encryptEntry = requestPubKey(this.clientConfig.getRsaPubKeyUrl(),
userName, false);
@@ -372,21 +370,21 @@ public class ProxyConfigManager extends Thread {
if (encryptEntry != null) {
encryptEntry.getRsaEncryptedKey();
synchronized (this) {
- if (this.userEncryConfigEntry == null) {
- this.userEncryConfigEntry = encryptEntry;
+ if (this.userEncryptConfigEntry == null) {
+ this.userEncryptConfigEntry = encryptEntry;
} else {
- encryptEntry = this.userEncryConfigEntry;
+ encryptEntry = this.userEncryptConfigEntry;
}
}
}
} else {
synchronized (this) {
- if (this.userEncryConfigEntry == null ||
this.userEncryConfigEntry != encryptEntry) {
+ if (this.userEncryptConfigEntry == null ||
this.userEncryptConfigEntry != encryptEntry) {
storePubKeyEntry(encryptEntry);
encryptEntry.getRsaEncryptedKey();
- this.userEncryConfigEntry = encryptEntry;
+ this.userEncryptConfigEntry = encryptEntry;
} else {
- encryptEntry = this.userEncryConfigEntry;
+ encryptEntry = this.userEncryptConfigEntry;
}
}
}
@@ -410,10 +408,10 @@ public class ProxyConfigManager extends Thread {
return;
}
synchronized (this) {
- if (this.userEncryConfigEntry == null || this.userEncryConfigEntry
!= encryptConfigEntry) {
+ if (this.userEncryptConfigEntry == null ||
this.userEncryptConfigEntry != encryptConfigEntry) {
storePubKeyEntry(encryptConfigEntry);
encryptConfigEntry.getRsaEncryptedKey();
- this.userEncryConfigEntry = encryptConfigEntry;
+ this.userEncryptConfigEntry = encryptConfigEntry;
}
}
return;
@@ -421,7 +419,7 @@ public class ProxyConfigManager extends Thread {
private EncryptConfigEntry getStoredPubKeyEntry(String userName) {
if (StringUtils.isBlank(userName)) {
- LOGGER.warn(" userName(" + userName + ") is not available");
+ logger.warn(" userName(" + userName + ") is not available");
return null;
}
EncryptConfigEntry entry;
@@ -441,7 +439,7 @@ public class ProxyConfigManager extends Thread {
return null;
}
} catch (Throwable e1) {
- LOGGER.error("Read " + userName + " stored PubKeyEntry error ",
e1);
+ logger.error("Read " + userName + " stored PubKeyEntry error ",
e1);
return null;
} finally {
if (fis != null) {
@@ -473,7 +471,7 @@ public class ProxyConfigManager extends Thread {
p.flush();
// p.close();
} catch (Throwable e) {
- LOGGER.error("store EncryptConfigEntry " + entry.toString() + "
exception ", e);
+ logger.error("store EncryptConfigEntry " + entry.toString() + "
exception ", e);
e.printStackTrace();
} finally {
if (fos != null) {
@@ -508,7 +506,7 @@ public class ProxyConfigManager extends Thread {
private EncryptConfigEntry requestPubKey(String pubKeyUrl, String
userName, boolean needGet) {
if (StringUtils.isBlank(userName)) {
- LOGGER.error("Queried userName is null!");
+ logger.error("Queried userName is null!");
return null;
}
List<BasicNameValuePair> params = new ArrayList<BasicNameValuePair>();
@@ -516,26 +514,26 @@ public class ProxyConfigManager extends Thread {
params.add(new BasicNameValuePair("username", userName));
String returnStr = requestConfiguration(pubKeyUrl, params);
if (StringUtils.isBlank(returnStr)) {
- LOGGER.info("No public key information returned from manager");
+ logger.info("No public key information returned from manager");
return null;
}
- JsonObject pubKeyConf = jsonParser.parse(returnStr).getAsJsonObject();
+ JsonObject pubKeyConf =
JsonParser.parseString(returnStr).getAsJsonObject();
if (pubKeyConf == null) {
- LOGGER.info("No public key information returned from manager");
+ logger.info("No public key information returned from manager");
return null;
}
if (!pubKeyConf.has("resultCode")) {
- LOGGER.info("Parse pubKeyConf failure: No resultCode key
information returned from manager");
+ logger.info("Parse pubKeyConf failure: No resultCode key
information returned from manager");
return null;
}
int resultCode = pubKeyConf.get("resultCode").getAsInt();
if (resultCode != 0) {
- LOGGER.info("query pubKeyConf failure, error code is " +
resultCode + ", errInfo is "
+ logger.info("query pubKeyConf failure, error code is " +
resultCode + ", errInfo is "
+ pubKeyConf.get("message").getAsString());
return null;
}
if (!pubKeyConf.has("resultData")) {
- LOGGER.info("Parse pubKeyConf failure: No resultData key
information returned from manager");
+ logger.info("Parse pubKeyConf failure: No resultData key
information returned from manager");
return null;
}
JsonObject resultData = pubKeyConf.get("resultData").getAsJsonObject();
@@ -566,7 +564,7 @@ public class ProxyConfigManager extends Thread {
throw new Exception("Read local proxyList File failure by " +
filePath + ", reason is " + e.getCause());
}
if (ObjectUtils.isEmpty(proxyCluster)) {
- LOGGER.warn("no proxyCluster configure from local file");
+ logger.warn("no proxyCluster configure from local file");
return null;
}
@@ -591,7 +589,7 @@ public class ProxyConfigManager extends Thread {
ArrayList<BasicNameValuePair> params = new
ArrayList<BasicNameValuePair>();
params.add(new BasicNameValuePair("ip", IpUtils.getLocalIp()));
params.add(new BasicNameValuePair("protocolType",
clientConfig.getProtocolType()));
- LOGGER.info("Begin to get configure from manager {}, param is {}",
url, params);
+ logger.info("Begin to get configure from manager {}, param is {}",
url, params);
String resultStr = requestConfiguration(url, params);
ProxyClusterConfig clusterConfig = gson.fromJson(resultStr,
ProxyClusterConfig.class);
@@ -606,7 +604,7 @@ public class ProxyConfigManager extends Thread {
private ProxyConfigEntry getProxyConfigEntry(DataProxyNodeResponse
proxyCluster) {
List<DataProxyNodeInfo> nodeList = proxyCluster.getNodeList();
if (CollectionUtils.isEmpty(nodeList)) {
- LOGGER.error("dataproxy nodeList is empty in
DataProxyNodeResponse!");
+ logger.error("dataproxy nodeList is empty in
DataProxyNodeResponse!");
return null;
}
Map<String, HostInfo> hostMap = formatHostInfoMap(nodeList);
@@ -644,133 +642,73 @@ public class ProxyConfigManager extends Thread {
}
private Map<String, HostInfo> formatHostInfoMap(List<DataProxyNodeInfo>
nodeList) {
+ HostInfo tmpHostInfo;
Map<String, HostInfo> hostMap = new HashMap<>();
for (DataProxyNodeInfo proxy : nodeList) {
if (ObjectUtils.isEmpty(proxy.getId()) ||
StringUtils.isEmpty(proxy.getIp()) || ObjectUtils
.isEmpty(proxy.getPort()) || proxy.getPort() < 0) {
- LOGGER.error("invalid proxy node, id:{}, ip:{}, port:{}",
proxy.getId(), proxy.getIp(),
+ logger.error("invalid proxy node, id:{}, ip:{}, port:{}",
proxy.getId(), proxy.getIp(),
proxy.getPort());
continue;
}
- String refId = proxy.getIp() + ":" + proxy.getPort();
- hostMap.put(refId, new HostInfo(refId, proxy.getIp(),
proxy.getPort()));
+ tmpHostInfo = new HostInfo(proxy.getIp(), proxy.getPort());
+ hostMap.put(tmpHostInfo.getReferenceName(), tmpHostInfo);
}
if (hostMap.isEmpty()) {
- LOGGER.error("Parse proxyList failure: address is empty for
response from manager!");
+ logger.error("Parse proxyList failure: address is empty for
response from manager!");
return null;
}
return hostMap;
}
- private String updateUrl(String url, int tryIdx, String
localManagerIpList) {
- if (tryIdx == 0) {
- return url;
- }
-
- int headerIdx = url.indexOf("://");
- if (headerIdx == -1) {
- return null;
- }
- String header = "";
- header = url.substring(0, headerIdx + 3);
- String tmpUrl = url.substring(headerIdx + 3);
- int tailerIdx = tmpUrl.indexOf("/");
- if (tailerIdx == -1) {
- return null;
- }
- String tailer = "";
- tailer = tmpUrl.substring(tailerIdx);
- String[] managerIps = localManagerIpList.split(",");
- String currentManagerIp = "";
- int idx = 1;
- for (String managerIp : managerIps) {
- if (idx++ == tryIdx) {
- currentManagerIp = managerIp;
- break;
- }
- }
- if (!currentManagerIp.equals("")) {
- return header + currentManagerIp + ":" +
clientConfig.getManagerPort() + tailer;
- }
- return null;
- }
-
/* Request new configurations from Manager. */
private String requestConfiguration(String url, List<BasicNameValuePair>
params) {
if (StringUtils.isBlank(url)) {
- LOGGER.error("request url is null");
+ logger.error("request url is null");
return null;
}
- // get local managerIpList
- String localManagerIps = "";
- int tryIdx = 0;
- while (true) {
- HttpPost httpPost = null;
- String returnStr = null;
- HttpParams myParams = new BasicHttpParams();
- HttpConnectionParams.setConnectionTimeout(myParams, 10000);
- HttpConnectionParams.setSoTimeout(myParams,
clientConfig.getManagerSocketTimeout());
- CloseableHttpClient httpClient = null;
- if (this.clientConfig.isRequestByHttp()) {
- httpClient = new DefaultHttpClient(myParams);
- } else {
- try {
- httpClient = getCloseableHttpClient(params);
- } catch (Throwable eHttps) {
- LOGGER.error("Create Https cliet failure, error 1 is ",
eHttps);
- eHttps.printStackTrace();
- return null;
- }
- }
-
- if (!clientConfig.isEnableSaveManagerVIps() && tryIdx > 0) {
+ HttpPost httpPost = null;
+ HttpParams myParams = new BasicHttpParams();
+ HttpConnectionParams.setConnectionTimeout(myParams, 10000);
+ HttpConnectionParams.setSoTimeout(myParams,
clientConfig.getManagerSocketTimeout());
+ CloseableHttpClient httpClient;
+ if (this.clientConfig.isRequestByHttp()) {
+ httpClient = new DefaultHttpClient(myParams);
+ } else {
+ try {
+ httpClient = getCloseableHttpClient(params);
+ } catch (Throwable eHttps) {
+ logger.error("Create Https cliet failure, error 1 is ",
eHttps);
+ eHttps.printStackTrace();
return null;
}
- // change url's manager host port when occur error
- url = updateUrl(url, tryIdx, localManagerIps);
- if (url == null) {
- return null;
+ }
+ logger.info("Request url : {}, params : {}", url, params);
+ try {
+ httpPost = new HttpPost(url);
+ httpPost.addHeader(BasicAuth.BASIC_AUTH_HEADER,
+
BasicAuth.genBasicAuthCredential(clientConfig.getAuthSecretId(),
+ clientConfig.getAuthSecretKey()));
+ UrlEncodedFormEntity urlEncodedFormEntity = new
UrlEncodedFormEntity(params, "UTF-8");
+ httpPost.setEntity(urlEncodedFormEntity);
+ HttpResponse response = httpClient.execute(httpPost);
+ String returnStr = EntityUtils.toString(response.getEntity());
+ if (StringUtils.isNotBlank(returnStr)
+ && response.getStatusLine().getStatusCode() ==
HttpStatus.SC_OK) {
+ logger.info("Get configure from manager is {}", returnStr);
+ return returnStr;
}
- tryIdx++;
-
- LOGGER.info("Request url : " + url + ", localManagerIps : " +
localManagerIps);
- try {
- httpPost = new HttpPost(url);
- httpPost.addHeader(BasicAuth.BASIC_AUTH_HEADER,
-
BasicAuth.genBasicAuthCredential(clientConfig.getAuthSecretId(),
- clientConfig.getAuthSecretKey()));
- UrlEncodedFormEntity urlEncodedFormEntity = new
UrlEncodedFormEntity(params, "UTF-8");
- httpPost.setEntity(urlEncodedFormEntity);
- HttpResponse response = httpClient.execute(httpPost);
- returnStr = EntityUtils.toString(response.getEntity());
- if (StringUtils.isNotBlank(returnStr)
- && response.getStatusLine().getStatusCode() ==
HttpStatus.SC_OK) {
- LOGGER.info("Get configure from manager is " + returnStr);
- return returnStr;
- }
-
- if (!clientConfig.isRequestByHttp()) {
- return null;
- }
- } catch (Throwable e) {
- LOGGER.error("Connect Manager error, message: {}, url is {}",
e.getMessage(), url);
-
- if (!clientConfig.isRequestByHttp()) {
- return null;
- }
- // get localManagerIps
- localManagerIps = getLocalManagerIps();
- if (localManagerIps == null) {
- return null;
- }
- } finally {
- if (httpPost != null) {
- httpPost.releaseConnection();
- }
- if (httpClient != null) {
- httpClient.getConnectionManager().shutdown();
- }
+ return null;
+ } catch (Throwable e) {
+ logger.error("Connect Manager error, message: {}, url is {}",
e.getMessage(), url);
+ return null;
+ } finally {
+ if (httpPost != null) {
+ httpPost.releaseConnection();
+ }
+ if (httpClient != null) {
+ httpClient.getConnectionManager().shutdown();
}
}
}
@@ -803,36 +741,8 @@ public class ProxyConfigManager extends Thread {
return httpClient;
}
- private String getLocalManagerIps() {
- String localManagerIps;
- try {
- File localManagerIpsFile = new
File(clientConfig.getManagerIpLocalPath());
- if (localManagerIpsFile.exists()) {
- byte[] serialized;
- serialized =
FileUtils.readFileToByteArray(localManagerIpsFile);
- if (serialized == null) {
- LOGGER.error("Local managerIp file is empty, file path : "
- + clientConfig.getManagerIpLocalPath());
- return null;
- }
- localManagerIps = new String(serialized, "UTF-8");
- } else {
- if (!localManagerIpsFile.getParentFile().exists()) {
- localManagerIpsFile.getParentFile().mkdirs();
- }
- localManagerIps = "";
- LOGGER.error("Get local managerIpList not exist, file path : "
- + clientConfig.getManagerIpLocalPath());
- }
- } catch (Throwable t) {
- localManagerIps = "";
- LOGGER.error("Get local managerIpList occur exception,", t);
- }
- return localManagerIps;
- }
-
public void updateHashRing(List<HostInfo> newHosts) {
this.hashRing.updateNode(newHosts);
- LOGGER.debug("update hash ring {}",
hashRing.getVirtualNode2RealNode());
+ logger.debug("update hash ring {}",
hashRing.getVirtualNode2RealNode());
}
}
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/SyncMessageCallable.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/SyncMessageCallable.java
index 8e00ff8c3e..2f75226925 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/SyncMessageCallable.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/SyncMessageCallable.java
@@ -19,6 +19,7 @@ package org.apache.inlong.sdk.dataproxy.network;
import org.apache.inlong.sdk.dataproxy.codec.EncodeObject;
import org.apache.inlong.sdk.dataproxy.common.SendResult;
+import org.apache.inlong.sdk.dataproxy.utils.LogCounter;
import io.netty.channel.ChannelFuture;
import org.slf4j.Logger;
@@ -30,8 +31,8 @@ import java.util.concurrent.TimeUnit;
public class SyncMessageCallable implements Callable<SendResult> {
- private static final Logger logger = LoggerFactory
- .getLogger(SyncMessageCallable.class);
+ private static final Logger logger =
LoggerFactory.getLogger(SyncMessageCallable.class);
+ private static final LogCounter exptCnt = new LogCounter(10, 100000, 60 *
1000L);
private final NettyClient client;
private final CountDownLatch awaitLatch = new CountDownLatch(1);
@@ -55,13 +56,13 @@ public class SyncMessageCallable implements
Callable<SendResult> {
}
public SendResult call() throws Exception {
- // TODO Auto-generated method stub
try {
ChannelFuture channelFuture = client.write(encodeObject);
awaitLatch.await(timeout, timeUnit);
- } catch (Exception e) {
- logger.error("SendResult call", e);
- e.printStackTrace();
+ } catch (Throwable ex) {
+ if (exptCnt.shouldPrint()) {
+ logger.warn("SyncMessageCallable write data throw exception",
ex);
+ }
return SendResult.UNKOWN_ERROR;
}
return message;
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/TimeScanObject.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/TimeScanObject.java
index 01d36c23fb..19291e4336 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/TimeScanObject.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/TimeScanObject.java
@@ -22,8 +22,8 @@ import java.util.concurrent.atomic.AtomicLong;
public class TimeScanObject {
- private AtomicInteger count = new AtomicInteger(0);
- private AtomicLong time = new AtomicLong(0);
+ private final AtomicInteger count = new AtomicInteger(0);
+ private final AtomicLong time = new AtomicLong(0);
public TimeScanObject() {
this.count.set(0);
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/threads/ManagerFetcherThread.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/threads/ManagerFetcherThread.java
deleted file mode 100644
index 1424c3cb8f..0000000000
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/threads/ManagerFetcherThread.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.sdk.dataproxy.threads;
-
-import org.apache.inlong.sdk.dataproxy.ProxyClientConfig;
-import org.apache.inlong.sdk.dataproxy.utils.ServiceDiscoveryUtils;
-
-import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.TimeUnit;
-
-/**
- * manager fetch thread
- */
-public class ManagerFetcherThread extends Thread {
-
- private final Logger logger =
LoggerFactory.getLogger(ManagerFetcherThread.class);
- private volatile boolean isShutdown;
- private final ProxyClientConfig proxyClientConfig;
-
- public ManagerFetcherThread(ProxyClientConfig proxyClientConfig) {
- isShutdown = false;
- this.proxyClientConfig = proxyClientConfig;
- this.setDaemon(true);
- this.setName("ManagerFetcherThread");
- }
-
- public void shutdown() {
- logger.info("Begin to shutdown ManagerFetcherThread.");
- isShutdown = true;
- }
-
- @Override
- public void run() {
- logger.info("ManagerFetcherThread Thread=" +
Thread.currentThread().getId() + " started !");
- while (!isShutdown) {
- try {
- String managerIpList =
ServiceDiscoveryUtils.getManagerIpList(proxyClientConfig);
- if (StringUtils.isBlank(managerIpList)) {
- logger.error("ManagerFetcher get managerIpList is blank.");
- } else {
-
ServiceDiscoveryUtils.updateManagerInfo2Local(managerIpList,
- proxyClientConfig.getManagerIpLocalPath());
- }
- TimeUnit.MILLISECONDS.sleep((long)
proxyClientConfig.getProxyUpdateIntervalMinutes() * 60 * 1000);
- } catch (Throwable e) {
- logger.error("ManagerFetcher get or save managerIpList occur
error,", e);
- }
- }
- }
-}
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/EventLoopUtil.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/EventLoopUtil.java
index bf26d3fc59..eb99c43b9e 100644
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/EventLoopUtil.java
+++
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/EventLoopUtil.java
@@ -53,12 +53,11 @@ public class EventLoopUtil {
} else if (!enableBusyWait) {
return new EpollEventLoopGroup(nThreads, threadFactory);
} else {
- EpollEventLoopGroup eventLoopGroup = new
EpollEventLoopGroup(nThreads, threadFactory, () -> {
+ return new EpollEventLoopGroup(nThreads, threadFactory, () -> {
return (selectSupplier, hasTasks) -> {
return -3;
};
});
- return eventLoopGroup;
}
}
diff --git
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/ServiceDiscoveryUtils.java
b/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/ServiceDiscoveryUtils.java
deleted file mode 100644
index 6352e25308..0000000000
---
a/inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/utils/ServiceDiscoveryUtils.java
+++ /dev/null
@@ -1,279 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.sdk.dataproxy.utils;
-
-import org.apache.inlong.common.util.BasicAuth;
-import org.apache.inlong.sdk.dataproxy.ProxyClientConfig;
-import org.apache.inlong.sdk.dataproxy.network.IpUtils;
-
-import com.google.gson.JsonArray;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonParser;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.http.Header;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.config.RequestConfig;
-import org.apache.http.client.entity.UrlEncodedFormEntity;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.DefaultHttpClient;
-import org.apache.http.impl.client.HttpClients;
-import org.apache.http.message.BasicHeader;
-import org.apache.http.message.BasicNameValuePair;
-import org.apache.http.params.BasicHttpParams;
-import org.apache.http.params.HttpConnectionParams;
-import org.apache.http.params.HttpParams;
-import org.apache.http.ssl.SSLContexts;
-import org.apache.http.util.EntityUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.net.ssl.SSLContext;
-
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.nio.charset.StandardCharsets;
-import java.security.SecureRandom;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Utils for service discovery
- */
-public class ServiceDiscoveryUtils {
-
- private static final Logger log =
LoggerFactory.getLogger(ServiceDiscoveryUtils.class);
-
- private static final String GET_MANAGER_IP_LIST_API =
"/inlong/manager/openapi/agent/getManagerIpList";
- private static String latestManagerIPList = "";
- private static String arraySed = ",";
-
- /**
- * Get Inlong-Manager IP list from the given proxy client config
- */
- public static String getManagerIpList(ProxyClientConfig clientConfig) {
- String managerAddress = clientConfig.getManagerAddress();
- if (StringUtils.isBlank(managerAddress)) {
- log.error("ServiceDiscovery get managerIpList but managerAddress
is blank, just return");
- return null;
- }
-
- String managerIpList = getManagerIpListByHttp(managerAddress,
clientConfig);
- if (StringUtils.isNotBlank(managerIpList)) {
- latestManagerIPList = managerIpList;
- return managerIpList;
- }
-
- log.error("ServiceDiscovery get managerIpList from {} occur error, try
to get from latestManagerIPList",
- managerAddress);
-
- String[] managerIps = latestManagerIPList.split(arraySed);
- if (managerIps.length > 0) {
- for (String managerIp : managerIps) {
- if (StringUtils.isBlank(managerIp)) {
- log.error("ServiceDiscovery managerIp is null,
latestManagerIPList is {}", latestManagerIPList);
- continue;
- }
-
- String currentAddress = managerIp + ":" +
clientConfig.getManagerPort();
- managerIpList = getManagerIpListByHttp(currentAddress,
clientConfig);
- if (StringUtils.isBlank(managerIpList)) {
- log.error("ServiceDiscovery get latestManagerIPList from
{} but got nothing, will try next ip",
- managerIp);
- continue;
- }
- latestManagerIPList = managerIpList;
- return managerIpList;
- }
- } else {
- log.error("ServiceDiscovery latestManagerIpList {} format error,
or not contain ip", latestManagerIPList);
- }
-
- String existedIpList =
getLocalManagerIpList(clientConfig.getManagerIpLocalPath());
- if (StringUtils.isNotBlank(existedIpList)) {
- String[] existedIps = existedIpList.split(arraySed);
- if (existedIps.length > 0) {
- for (String existedIp : existedIps) {
- if (StringUtils.isBlank(existedIp)) {
- log.error("ServiceDiscovery get illegal format ipList
from local file, "
- + "exist ip is empty, managerIpList is {},
local file is {}",
- existedIpList,
clientConfig.getManagerIpLocalPath());
- continue;
- }
-
- String currentAddress = existedIp + ":" +
clientConfig.getManagerPort();
- managerIpList = getManagerIpListByHttp(currentAddress,
clientConfig);
- if (StringUtils.isBlank(managerIpList)) {
- log.error("ServiceDiscovery get {} from local file {}
but got nothing, will try next ip",
- existedIp,
clientConfig.getManagerIpLocalPath());
- continue;
- }
- latestManagerIPList = managerIpList;
- return managerIpList;
- }
- } else {
- log.error("ServiceDiscovery get illegal format ipList from
local file, "
- + "exist ip is empty, managerIpList is {}, local file
is {}",
- existedIpList, clientConfig.getManagerIpLocalPath());
- }
- } else {
- log.error("ServiceDiscovery get empty ipList from local file {}",
clientConfig.getManagerIpLocalPath());
- }
-
- return managerIpList;
- }
-
- /**
- * Get Inlong-Manager IP list from the given managerIp and proxy client
config
- */
- public static String getManagerIpListByHttp(String managerIp,
ProxyClientConfig proxyClientConfig) {
- String url = managerIp + GET_MANAGER_IP_LIST_API;
- ArrayList<BasicNameValuePair> params = new
ArrayList<BasicNameValuePair>();
- params.add(new BasicNameValuePair("operation", "query"));
- params.add(new BasicNameValuePair("username",
proxyClientConfig.getUserName()));
-
- log.info("Begin to get configure from manager {}, param is {}", url,
params);
- CloseableHttpClient httpClient;
- HttpParams myParams = new BasicHttpParams();
- HttpConnectionParams.setConnectionTimeout(myParams,
proxyClientConfig.getManagerConnectionTimeout());
- HttpConnectionParams.setSoTimeout(myParams,
proxyClientConfig.getManagerSocketTimeout());
- if (proxyClientConfig.isRequestByHttp()) {
- httpClient = new DefaultHttpClient(myParams);
- } else {
- try {
- ArrayList<Header> headers = new ArrayList<>();
- for (BasicNameValuePair paramItem : params) {
- headers.add(new BasicHeader(paramItem.getName(),
paramItem.getValue()));
- }
- RequestConfig requestConfig = RequestConfig.custom()
-
.setConnectTimeout(10000).setSocketTimeout(30000).build();
- SSLContext sslContext = SSLContexts.custom().build();
- SSLConnectionSocketFactory sslsf = new
SSLConnectionSocketFactory(sslContext,
- new String[]{"TLSv1"}, null,
-
SSLConnectionSocketFactory.getDefaultHostnameVerifier());
- httpClient = HttpClients.custom().setDefaultHeaders(headers)
-
.setDefaultRequestConfig(requestConfig).setSSLSocketFactory(sslsf).build();
- } catch (Throwable t) {
- log.error("Create Https client failed: ", t);
- return null;
- }
- }
-
- HttpPost httpPost = null;
- try {
- httpPost = new HttpPost(url);
- if (proxyClientConfig.isNeedAuthentication()) {
- long timestamp = System.currentTimeMillis();
- int nonce = new
SecureRandom(String.valueOf(timestamp).getBytes()).nextInt(Integer.MAX_VALUE);
- httpPost.setHeader(BasicAuth.BASIC_AUTH_HEADER,
-
IpUtils.getAuthorizenInfo(proxyClientConfig.getUserName(),
- proxyClientConfig.getSecretKey(), timestamp,
nonce));
- }
- httpPost.setEntity(new UrlEncodedFormEntity(params));
- HttpResponse response = httpClient.execute(httpPost);
- String returnStr = EntityUtils.toString(response.getEntity());
- if (StringUtils.isNotBlank(returnStr) &&
response.getStatusLine().getStatusCode() == 200) {
- log.info("Get configure from manager is " + returnStr);
- JsonParser jsonParser = new JsonParser();
- JsonObject jb = jsonParser.parse(returnStr).getAsJsonObject();
- if (jb == null) {
- log.warn("ServiceDiscovery updated manager ip failed,
returnStr = {} jb is "
- + "null ", returnStr, jb);
- return null;
- }
- JsonArray retData = jb.get("data").getAsJsonArray();
- List<String> managerIpList = new ArrayList<>();
- for (JsonElement datum : retData) {
- JsonObject record = datum.getAsJsonObject();
- managerIpList.add(record.get("ip").getAsString());
- }
- if (managerIpList.isEmpty()) {
- return null;
- }
- String strIPs = String.join(",", managerIpList);
- log.info("ServiceDiscovery updated manager ip success, ip : "
+ strIPs + ", retStr : " + returnStr);
- return strIPs;
- }
- return null;
- } catch (Throwable t) {
- log.error("Connect Manager error: ", t);
- return null;
- } finally {
- if (httpPost != null) {
- httpPost.releaseConnection();
- }
- if (httpClient != null) {
- httpClient.getConnectionManager().shutdown();
- }
- }
- }
-
- /**
- * Get Inlong-Manager IP list from local path
- */
- public static String getLocalManagerIpList(String localPath) {
- log.info("ServiceDiscovery start loading config from file {} ...",
localPath);
- String newestIp = null;
- try {
- File managerIpListFile = new File(localPath);
- if (!managerIpListFile.exists()) {
- log.info("ServiceDiscovery not found local groupIdInfo file
from {}", localPath);
- return null;
- }
- byte[] serialized =
FileUtils.readFileToByteArray(managerIpListFile);
- if (serialized == null) {
- return null;
- }
- newestIp = new String(serialized, StandardCharsets.UTF_8);
- log.info("ServiceDiscovery get manager ip list from local success,
result is: {}", newestIp);
- } catch (IOException e) {
- log.error("ServiceDiscovery load manager config error: ", e);
- }
-
- return newestIp;
- }
-
- /**
- * Update Inlong-Manager info to local file
- */
- public static void updateManagerInfo2Local(String storeString, String
path) {
- if (StringUtils.isBlank(storeString)) {
- log.warn("ServiceDiscovery updateTdmInfo2Local error, configMap is
empty or managerIpList is blank");
- return;
- }
- File localPath = new File(path);
- if (!localPath.getParentFile().exists()) {
- localPath.getParentFile().mkdirs();
- }
-
- try (BufferedWriter writer = new BufferedWriter(
- new OutputStreamWriter(new FileOutputStream(localPath),
StandardCharsets.UTF_8))) {
- writer.write(storeString);
- writer.flush();
- } catch (IOException e) {
- log.error("ServiceDiscovery save manager config error: ", e);
- }
- }
-
-}
