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/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 7840029 [INLONG-1987][Improve] Add function comment information in
TubeMQ (#1988)
7840029 is described below
commit 78400294c870c0a42abd43c7111f7ebdb53570d8
Author: gosonzhang <[email protected]>
AuthorDate: Tue Dec 14 20:49:04 2021 +0800
[INLONG-1987][Improve] Add function comment information in TubeMQ (#1988)
---
.../tubemq/client/config/TubeClientConfig.java | 18 +++++++
.../client/config/TubeClientConfigUtils.java | 6 +++
.../tubemq/client/consumer/ClientSubInfo.java | 9 ++++
.../tubemq/client/consumer/MessageConsumer.java | 5 +-
.../client/consumer/MessageFetchManager.java | 7 ++-
.../tubemq/client/consumer/RmtDataCache.java | 8 ++-
.../org/apache/inlong/tubemq/corebase/Message.java | 34 +++++++-----
.../tubemq/corebase/policies/FlowCtrlItem.java | 6 +++
.../corebase/policies/FlowCtrlRuleHandler.java | 62 +++++++++++++---------
.../inlong/tubemq/corebase/utils/AddressUtils.java | 10 ++++
.../inlong/tubemq/corebase/utils/OpsSyncInfo.java | 5 ++
.../broker/metadata/BrokerMetadataManagerTest.java | 10 ++--
.../server/broker/utils/DataStoreUtilsTest.java | 4 +-
13 files changed, 133 insertions(+), 51 deletions(-)
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/config/TubeClientConfig.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/config/TubeClientConfig.java
index c567c37..337935c 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/config/TubeClientConfig.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/config/TubeClientConfig.java
@@ -364,6 +364,13 @@ public class TubeClientConfig {
this.sessionMaxAllowedDelayedMsgCount =
sessionMaxAllowedDelayedMsgCount;
}
+ /**
+ * Set authenticate information
+ *
+ * @param needAuthentic enable or disable authentication
+ * @param usrName the user name
+ * @param usrPassWord the password
+ */
public void setAuthenticInfo(boolean needAuthentic,
String usrName,
String usrPassWord) {
@@ -385,6 +392,12 @@ public class TubeClientConfig {
}
}
+ /**
+ * Set TLS information
+ *
+ * @param trustStorePath the trusted store path
+ * @param trustStorePassword the trusted store password
+ */
public void setTLSEnableInfo(String trustStorePath, String
trustStorePassword) {
// public void setTLSEnableInfo(String trustStorePath, String
trustStorePassword,
// boolean tlsTwoWayAuthEnable,String keyStorePath, String
keyStorePassword) throws Exception {
@@ -544,6 +557,11 @@ public class TubeClientConfig {
return masterInfo.equals(that.masterInfo);
}
+ /**
+ * Get the configured Json string information
+ *
+ * @return the configured Json string information
+ */
public String toJsonString() {
int num = 0;
String localAddress = null;
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/config/TubeClientConfigUtils.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/config/TubeClientConfigUtils.java
index c2e04f4..a9005eb 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/config/TubeClientConfigUtils.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/config/TubeClientConfigUtils.java
@@ -22,6 +22,12 @@ import org.apache.inlong.tubemq.corerpc.RpcConstants;
public class TubeClientConfigUtils {
+ /**
+ * Get RPC configure by client configure information
+ *
+ * @param tubeClientConfig the client configure
+ * @param isSingleSession whether single session factory
+ */
public static RpcConfig getRpcConfigByClientConfig(final TubeClientConfig
tubeClientConfig,
boolean
isSingleSession) {
RpcConfig config = new RpcConfig();
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/ClientSubInfo.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/ClientSubInfo.java
index 3348801..edd716c 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/ClientSubInfo.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/ClientSubInfo.java
@@ -139,6 +139,15 @@ public class ClientSubInfo {
this.subscribedTime = System.currentTimeMillis();
}
+ /**
+ * Set Bound Consumption information
+ *
+ * @param sessionKey consume session key
+ * @param sourceCount the client count of consume group
+ * @param isSelectBig whether select a bigger data If there is reset
conflict
+ * @param partOffsetMap the map of partitionKey and bootstrap offset
+ *
+ */
public void setRequireBound(final String sessionKey,
final int sourceCount,
final boolean isSelectBig,
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageConsumer.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageConsumer.java
index df98490..6d650b7 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageConsumer.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageConsumer.java
@@ -47,7 +47,7 @@ public interface MessageConsumer extends Shutdownable {
* consume data until the partition is unfrozen or
* rebalanced to other clients in the same group
*
- * @return void
+ * @param partitionKeys The partitionKey list that needs to be frozen
*/
void freezePartitions(List<String> partitionKeys) throws
TubeClientException;
@@ -55,7 +55,7 @@ public interface MessageConsumer extends Shutdownable {
* unfreeze frozen partitions, the specified partition will
* resume data consumption until the partition is frozen again
*
- * @return void
+ * @param partitionKeys The partitionKey list that needs to be unfrozen
*/
void unfreezePartitions(List<String> partitionKeys) throws
TubeClientException;
@@ -63,7 +63,6 @@ public interface MessageConsumer extends Shutdownable {
* unfreeze all frozen partitions, the unfreeze partition will
* resume data consumption until the partition is frozen again
*
- * @return void
*/
void relAllFrozenPartitions();
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageFetchManager.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageFetchManager.java
index c2202d2..b2b7868 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageFetchManager.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/MessageFetchManager.java
@@ -53,7 +53,7 @@ public class MessageFetchManager {
/**
* Start a worker pool to fetch the message.
*
- * @throws TubeClientException
+ * @throws TubeClientException Exception thrown
*/
public void startFetchWorkers() throws TubeClientException {
this.pushConsumer.getBaseConsumer().checkClientRunning();
@@ -95,6 +95,11 @@ public class MessageFetchManager {
return this.managerStatus.get() == 0;
}
+ /**
+ * Stop fetch worker threads
+ *
+ * @param onlySetStatus Whether to only set the state without stopping
the operation
+ */
public void stopFetchWorkers(boolean onlySetStatus) throws
InterruptedException {
if (onlySetStatus) {
if (this.managerStatus.get() == 0) {
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java
index b7d111c..34619ee 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java
@@ -315,6 +315,12 @@ public class RmtDataCache implements Closeable {
return this.groupFlowCtrlRuleHandler.getFlowCtrlId();
}
+ /**
+ * store topic meta information
+ *
+ * @param curTopicMetaInfoId the topic meta information id
+ * @param curMetaInfoSet the topic meta information
+ */
public void storeTopicMetaInfo(long curTopicMetaInfoId, List<String>
curMetaInfoSet) {
if (curTopicMetaInfoId < 0
|| curTopicMetaInfoId == this.topicMetaInfoId.get()) {
@@ -580,7 +586,7 @@ public class RmtDataCache implements Closeable {
* @param limitDlt max offset of the data fetch
* @param curDataDlt the offset of current data fetch
* @param isRequireSlow if the server requires slow down
- * @param maxOffset partiton current max offset
+ * @param maxOffset partition current max offset
*/
public void setPartitionContextInfo(String partitionKey, long currOffset,
int reqProcType, int errCode,
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/Message.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/Message.java
index f8682de..1ab206c 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/Message.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/Message.java
@@ -54,8 +54,8 @@ public class Message implements Serializable {
/**
* init with topic and data body
*
- * @param topic
- * @param data
+ * @param topic topic name
+ * @param data message content
*/
public Message(final String topic, final byte[] data) {
super();
@@ -66,11 +66,11 @@ public class Message implements Serializable {
/**
* init with indexId topic data attribute flag
*
- * @param indexId
- * @param topic
- * @param data
- * @param attribute
- * @param flag
+ * @param indexId index id
+ * @param topic topic name
+ * @param data message content
+ * @param attribute message attribute
+ * @param flag message flag
*/
protected Message(long indexId, String topic, byte[] data, String
attribute, int flag) {
this.indexId = indexId;
@@ -105,8 +105,12 @@ public class Message implements Serializable {
}
/**
- * @param msgType
- * @param msgTime
+ * Set the filtering items that need to be filtered and the message
forwarding time
+ *
+ * @param msgType the ID of the stream that needs to be filtered,
+ * if not required, set null
+ * @param msgTime the time when the message was sent, the format is
yyyyMMddHHmm,
+ * if not required, set null
*/
public void putSystemHeader(final String msgType, final String msgTime) {
this.msgType = null;
@@ -194,8 +198,10 @@ public class Message implements Serializable {
}
/**
- * @param keyVal
- * @return
+ * Get key's value from attribute by key name
+ *
+ * @param keyVal the key name
+ * @return the key's value of the key name stored in attribute
*/
public String getAttrValue(final String keyVal) {
if (TStringUtils.isBlank(keyVal)) {
@@ -219,8 +225,10 @@ public class Message implements Serializable {
}
/**
- * @param keyVal
- * @param valueVal
+ * Store key and value in attribute
+ *
+ * @param keyVal the key name
+ * @param valueVal the key value
*/
public void setAttrKeyVal(final String keyVal, final String valueVal) {
if (TStringUtils.isBlank(keyVal)) {
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlItem.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlItem.java
index b0cb889..555ed2b 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlItem.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlItem.java
@@ -146,6 +146,12 @@ public class FlowCtrlItem {
return zeroCnt;
}
+ /**
+ * Build json result string buffer
+ *
+ * @param sBuilder the string buffer
+ * @return the result content
+ */
public StringBuilder toJsonString(final StringBuilder sBuilder) {
switch (this.type) {
case 1:
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlRuleHandler.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlRuleHandler.java
index 4fe20f9..8f26221 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlRuleHandler.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlRuleHandler.java
@@ -87,10 +87,12 @@ public class FlowCtrlRuleHandler {
}
/**
- * @param qyrPriorityId
- * @param flowCtrlId
- * @param flowCtrlInfo
- * @throws Exception
+ * Parse flow control information and update stored cached old content
+ *
+ * @param qyrPriorityId query priority id
+ * @param flowCtrlId flow control information id
+ * @param flowCtrlInfo flow control information content
+ * @throws Exception Exception thrown
*/
public void updateFlowCtrlInfo(final int qyrPriorityId,
final long flowCtrlId,
@@ -126,8 +128,10 @@ public class FlowCtrlRuleHandler {
}
/**
- * @param lastDataDlt
- * @return FlowCtrlResult
+ * Get current data lag limit strategy
+ *
+ * @param lastDataDlt current consumption lag of data
+ * @return FlowCtrlResult current flow control policy
*/
public FlowCtrlResult getCurDataLimit(long lastDataDlt) {
Calendar rightNow = Calendar.getInstance(timeZone);
@@ -171,7 +175,7 @@ public class FlowCtrlRuleHandler {
}
/**
- * initial statis data
+ * Initial data limit statistics
*/
private void initialStatisData() {
initialDataLimitStatisInfo();
@@ -180,7 +184,7 @@ public class FlowCtrlRuleHandler {
}
/**
- * initial data limit statis info
+ * Initial data limit statistics
*/
private void initialDataLimitStatisInfo() {
List<FlowCtrlItem> flowCtrlItemList = this.flowCtrlRuleSet.get(0);
@@ -256,9 +260,11 @@ public class FlowCtrlRuleHandler {
}
/**
- * @param msgZeroCnt
- * @param rcmVal
- * @return
+ * Get the current fetch frequency limit strategy
+ *
+ * @param msgZeroCnt the continuous consumption count without messages
+ * @param rcmVal the default frequency limit value
+ * @return the required frequency limit value
*/
public int getCurFreqLimitTime(int msgZeroCnt, int rcmVal) {
if (msgZeroCnt < this.minZeroCnt.get()) {
@@ -287,7 +293,8 @@ public class FlowCtrlRuleHandler {
}
/**
- * @param qryPriorityId
+ * Set query priority id value
+ * @param qryPriorityId need to set value
*/
public void setQryPriorityId(int qryPriorityId) {
this.qryPriorityId.set(qryPriorityId);
@@ -311,9 +318,10 @@ public class FlowCtrlRuleHandler {
/**
* Parse FlowCtrlInfo value
- * @param flowCtrlInfo flowCtrlInfo json value
- * @return parse result
- * @throws Exception
+ *
+ * @param flowCtrlInfo flowCtrlInfo json value
+ * @return parse result
+ * @throws Exception Exception thrown
*/
public Map<Integer, List<FlowCtrlItem>> parseFlowCtrlInfo(final String
flowCtrlInfo)
throws Exception {
@@ -393,8 +401,8 @@ public class FlowCtrlRuleHandler {
* @param recordNo record no
* @param typeVal type value
* @param jsonObject record json value
- * @return parsed result
- * @throws Exception
+ * @return parsed result
+ * @throws Exception Exception thrown
*/
private List<FlowCtrlItem> parseDataLimit(int recordNo, int typeVal,
JsonObject jsonObject) throws
Exception {
@@ -501,8 +509,8 @@ public class FlowCtrlRuleHandler {
* @param recordNo record no
* @param typeVal type value
* @param jsonObject record json value
- * @return parsed result
- * @throws Exception
+ * @return parsed result
+ * @throws Exception Exception thrown
*/
private List<FlowCtrlItem> parseFreqLimit(int recordNo, int typeVal,
JsonObject jsonObject) throws
Exception {
@@ -580,11 +588,12 @@ public class FlowCtrlRuleHandler {
/**
* Parse low frequent fetch count
+ *
* @param recordNo record no
* @param typeVal type value
* @param jsonObject record json value
- * @return parsed result
- * @throws Exception
+ * @return parsed result
+ * @throws Exception Exception thrown
*/
private List<FlowCtrlItem> parseLowFetchLimit(int recordNo, int typeVal,
JsonObject jsonObject)
throws Exception {
@@ -700,12 +709,13 @@ public class FlowCtrlRuleHandler {
/**
* Parse time information
+ *
* @param ruleObject rule value object
- * @param fieldName field name
- * @param itemNo rule no
- * @param recordNo record no
- * @return parse result
- * @throws Exception
+ * @param fieldName field name
+ * @param itemNo rule no
+ * @param recordNo record no
+ * @return parse result
+ * @throws Exception Exception thrown
*/
private int validAndGetTimeValue(JsonObject ruleObject, String fieldName,
int itemNo, int recordNo) throws
Exception {
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/AddressUtils.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/AddressUtils.java
index b846f95..970d7cd 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/AddressUtils.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/AddressUtils.java
@@ -138,6 +138,11 @@ public class AddressUtils {
}
}
+ /**
+ * Get local IP V4 address
+ *
+ * @return the local IP V4 address
+ */
public static String getIPV4LocalAddress() {
if (localIPAddress != null) {
return localIPAddress;
@@ -175,6 +180,11 @@ public class AddressUtils {
.append(" in local networkInterfaces!").toString());
}
+ /**
+ * Get the local IP V4 address from the designated NetworkInterface
+ *
+ * @return the local IP V4 address
+ */
public static String getIPV4LocalAddress(String defEthName) {
boolean foundNetInter = false;
try {
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/OpsSyncInfo.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/OpsSyncInfo.java
index 4eb9dc6..4e86c73 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/OpsSyncInfo.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/OpsSyncInfo.java
@@ -35,6 +35,11 @@ public class OpsSyncInfo {
}
+ /**
+ * Update Ops task information
+ *
+ * @param opsTaskInfo the ops task information sent from Master
+ */
public void updOpsSyncInfo(ClientMaster.OpsTaskInfo opsTaskInfo) {
if (opsTaskInfo == null) {
return;
diff --git
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerMetadataManagerTest.java
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerMetadataManagerTest.java
index 906139e..7e9dd7e 100644
---
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerMetadataManagerTest.java
+++
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerMetadataManagerTest.java
@@ -20,8 +20,8 @@ package org.apache.inlong.tubemq.server.broker.metadata;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import org.junit.Assert;
import org.junit.Test;
-import org.springframework.util.Assert;
/***
* BrokerMetadataManage test
@@ -44,13 +44,13 @@ public class BrokerMetadataManagerTest {
newBrokerDefMetaConfInfo, newTopicMetaConfInfoList, true, new
StringBuilder());
// get topic custom config.
long count = brokerMetadataManager.getNumPartitions("topic2");
- Assert.isTrue(count == 4);
+ Assert.assertEquals(count, 4);
// add topic custom config.
newTopicMetaConfInfoList.add("topic3:6:true:true:1000:10000:0,0,6:delete,168h:1:1000:1024:1000:1000:1");
brokerMetadataManager.updateBrokerTopicConfigMap(0L, 1,
newBrokerDefMetaConfInfo, newTopicMetaConfInfoList, true, new
StringBuilder());
count = brokerMetadataManager.getNumPartitions("topic3");
- Assert.isTrue(count == 6);
+ Assert.assertEquals(count, 6);
}
@Test
@@ -64,12 +64,12 @@ public class BrokerMetadataManagerTest {
brokerMetadataManager.updateBrokerTopicConfigMap(0L, 0,
newBrokerDefMetaConfInfo, newTopicMetaConfInfoList, true, new
StringBuilder());
Map<String, TopicMetadata> topicMetadataMap =
brokerMetadataManager.getRemovedTopicConfigMap();
- Assert.isTrue(topicMetadataMap.size() == 0);
+ Assert.assertEquals(topicMetadataMap.size(), 0);
List<String> rmvTopics = new LinkedList<>();
rmvTopics.add("topic2:4:true:true:1000:10000:0,0,6:delete,168h:1:1000:1024:1000:1000:1");
// update topic custom config.
brokerMetadataManager.updateBrokerRemoveTopicMap(true, rmvTopics, new
StringBuilder());
topicMetadataMap = brokerMetadataManager.getRemovedTopicConfigMap();
- Assert.isTrue(topicMetadataMap.size() == 1);
+ Assert.assertEquals(topicMetadataMap.size(), 1);
}
}
diff --git
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtilsTest.java
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtilsTest.java
index 45ddf29..c240064 100644
---
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtilsTest.java
+++
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtilsTest.java
@@ -18,8 +18,8 @@
package org.apache.inlong.tubemq.server.broker.utils;
import java.nio.ByteBuffer;
+import org.junit.Assert;
import org.junit.Test;
-import org.springframework.util.Assert;
/***
* DataStoreUtils test.
@@ -34,6 +34,6 @@ public class DataStoreUtilsTest {
int offset = 0;
int val = DataStoreUtils.getInt(offset, data);
// get int by DataStoreUtils
- Assert.isTrue(val == 123);
+ Assert.assertEquals(val, 123);
}
}
