This is an automated email from the ASF dual-hosted git repository.
healchow 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 7a0b61f [INLONG-2279][Improve] Supplement TubeMQ's Javadoc
information (#2280)
7a0b61f is described below
commit 7a0b61f634f7b25f0fc21c55f0de21b6dba6470f
Author: gosonzhang <[email protected]>
AuthorDate: Sun Jan 23 16:14:10 2022 +0800
[INLONG-2279][Improve] Supplement TubeMQ's Javadoc information (#2280)
---
.../inlong/tubemq/corebase/cluster/Partition.java | 6 +-
.../tubemq/corebase/utils/DataConverterUtil.java | 4 +-
.../inlong/tubemq/corerpc/codec/PbEnDecoder.java | 2 +-
.../tubemq/corerpc/netty/NettyRpcServer.java | 2 +-
.../corerpc/netty/NettyProtocolEncoderTest.java | 2 +-
.../tubemq/corerpc/utils/TSSLEngineUtilTest.java | 2 +-
.../example/ClientBalanceConsumerExample.java | 13 ++
.../tubemq/example/MAMessageProducerExample.java | 21 ++-
.../tubemq/example/MessageProducerExample.java | 17 +-
.../tubemq/example/MessagePullConsumerExample.java | 19 ++-
.../example/MessagePullSetConsumerExample.java | 19 ++-
.../tubemq/example/MessagePushConsumerExample.java | 19 ++-
.../inlong/tubemq/example/MsgSendReceiveStats.java | 10 +-
.../inlong/tubemq/server/broker/BrokerConfig.java | 4 +-
.../tubemq/server/broker/BrokerServiceServer.java | 172 ++++++++++-----------
.../inlong/tubemq/server/broker/TubeBroker.java | 43 +++---
.../broker/exception/OffsetStoreException.java | 2 +-
.../server/broker/exception/StartupException.java | 2 +-
.../server/broker/metadata/BrokerDefMetadata.java | 2 +-
.../broker/metadata/BrokerMetadataManager.java | 10 +-
.../server/broker/metadata/MetadataManager.java | 2 +-
.../server/broker/metadata/TopicMetadata.java | 8 +-
.../server/broker/msgstore/MessageStore.java | 32 ++--
.../broker/msgstore/MessageStoreManager.java | 61 +++++---
.../server/broker/msgstore/StoreService.java | 2 +-
.../server/broker/msgstore/disk/FileSegment.java | 16 +-
.../broker/msgstore/disk/FileSegmentList.java | 14 +-
.../broker/msgstore/disk/GetMessageResult.java | 2 +-
.../broker/msgstore/disk/MsgFileStatisInfo.java | 2 +-
.../server/broker/msgstore/disk/MsgFileStore.java | 25 ++-
.../server/broker/msgstore/disk/Segment.java | 8 +-
.../server/broker/msgstore/disk/SegmentList.java | 2 +-
.../server/broker/msgstore/disk/SegmentType.java | 2 +-
.../broker/msgstore/mem/GetCacheMsgResult.java | 2 +-
.../broker/msgstore/mem/MsgMemStatisInfo.java | 20 +--
.../server/broker/msgstore/mem/MsgMemStore.java | 10 +-
.../server/broker/nodeinfo/ConsumerNodeInfo.java | 14 +-
.../server/broker/offset/DefaultOffsetManager.java | 39 ++---
.../server/broker/offset/OffsetRecordInfo.java | 6 +-
.../server/broker/offset/OffsetRecordService.java | 2 +-
.../tubemq/server/broker/offset/OffsetService.java | 2 +-
.../tubemq/server/broker/offset/RecordItem.java | 2 +-
.../tubemq/server/broker/stats/CountItem.java | 2 +-
.../server/broker/stats/GroupCountService.java | 2 +-
.../server/broker/utils/BrokerSamplePrint.java | 2 +-
.../tubemq/server/broker/utils/DataStoreUtils.java | 16 +-
.../server/broker/utils/DiskSamplePrint.java | 2 +-
.../server/broker/web/BrokerAdminServlet.java | 34 ++--
.../inlong/tubemq/server/broker/web/WebServer.java | 2 +-
.../server/common/paramcheck/PBParameterUtils.java | 22 ++-
.../nodemanage/nodeconsumer/ConsumeGroupInfo.java | 27 +++-
.../nodeconsumer/ConsumerEventManager.java | 9 +-
.../inlong/tubemq/server/master/utils/Chore.java | 6 +-
.../server/master/web/common/BaseResult.java | 91 ++++++++---
.../master/web/simplemvc/RequestDispatcher.java | 14 +-
.../web/simplemvc/conf/ConfigFileParser.java | 5 +
.../inlong/tubemq/server/tools/BdbGroupAdmin.java | 8 +-
.../inlong/tubemq/server/tools/CliUtils.java | 12 ++
.../tubemq/server/tools/StoreRepairAdmin.java | 11 ++
.../tubemq/server/tools/cli/CliAbstractBase.java | 2 +-
.../tubemq/server/tools/cli/CliBrokerAdmin.java | 7 +-
.../tubemq/server/tools/cli/CliConsumer.java | 13 +-
.../tubemq/server/tools/cli/CliProducer.java | 11 +-
.../broker/metadata/BrokerMetadataManagerTest.java | 2 +-
.../broker/msgstore/disk/FileSegmentListTest.java | 2 +-
.../broker/msgstore/disk/FileSegmentTest.java | 2 +-
.../broker/msgstore/mem/MsgMemStoreTest.java | 2 +-
.../server/broker/stats/GroupCountServiceTest.java | 2 +-
.../server/broker/utils/DataStoreUtilsTest.java | 2 +-
69 files changed, 579 insertions(+), 375 deletions(-)
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/cluster/Partition.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/cluster/Partition.java
index 71c49d4..4f658cb 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/cluster/Partition.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/cluster/Partition.java
@@ -51,10 +51,10 @@ public class Partition implements Comparable<Partition>,
Serializable {
this.builderPartitionStr();
}
- /********
+ /**
* create a Partition with full part info String
*
- * @param strPartInfo
+ * @param strPartInfo the partition information in string format
*/
public Partition(String strPartInfo) {
this.broker =
@@ -65,7 +65,7 @@ public class Partition implements Comparable<Partition>,
Serializable {
this.builderPartitionStr();
}
- /******
+ /**
* create a Partition with full broker and part info string
*
* @param broker
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/DataConverterUtil.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/DataConverterUtil.java
index aaf5a8d..39b2c5b 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/DataConverterUtil.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/DataConverterUtil.java
@@ -123,7 +123,7 @@ public class DataConverterUtil {
return topicList;
}
- /*********
+ /**
* convert string info to @link BrokerInfo
*
* @param strBrokerInfos return a BrokerInfo Map
@@ -143,7 +143,7 @@ public class DataConverterUtil {
return brokerInfoMap;
}
- /*********
+ /**
* convert string info to a map of TopicCondition TreeSet
*
* @param strTopicConditions return a map of TopicCondition TreeSet
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/codec/PbEnDecoder.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/codec/PbEnDecoder.java
index 036c9d5..9e394d3 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/codec/PbEnDecoder.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/codec/PbEnDecoder.java
@@ -24,7 +24,7 @@ import
org.apache.inlong.tubemq.corebase.protobuf.generated.ClientBroker;
import org.apache.inlong.tubemq.corebase.protobuf.generated.ClientMaster;
import org.apache.inlong.tubemq.corerpc.RpcConstants;
-/***
+/**
* PB corresponding method, service type codec util tools
*/
public class PbEnDecoder {
diff --git
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java
index ba8fcee..830f8e5 100644
---
a/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java
+++
b/inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java
@@ -60,7 +60,7 @@ import org.jboss.netty.handler.ssl.SslHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Netty Rpc Server
*/
public class NettyRpcServer implements ServiceRpcServer {
diff --git
a/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoderTest.java
b/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoderTest.java
index 59e0a76..44e0c2f 100644
---
a/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoderTest.java
+++
b/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/netty/NettyProtocolEncoderTest.java
@@ -25,7 +25,7 @@ import org.jboss.netty.buffer.ChannelBuffer;
import org.junit.Assert;
import org.junit.Test;
-/***
+/**
* NettyProtocolEncoder test.
*/
public class NettyProtocolEncoderTest {
diff --git
a/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/utils/TSSLEngineUtilTest.java
b/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/utils/TSSLEngineUtilTest.java
index 4650d95..d6e24bd 100644
---
a/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/utils/TSSLEngineUtilTest.java
+++
b/inlong-tubemq/tubemq-core/src/test/java/org/apache/inlong/tubemq/corerpc/utils/TSSLEngineUtilTest.java
@@ -23,7 +23,7 @@ import javax.net.ssl.SSLEngine;
import org.junit.Assert;
-/***
+/**
* TSSLEngineUtil test.
*/
public class TSSLEngineUtilTest {
diff --git
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/ClientBalanceConsumerExample.java
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/ClientBalanceConsumerExample.java
index c495234..fde2afa 100644
---
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/ClientBalanceConsumerExample.java
+++
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/ClientBalanceConsumerExample.java
@@ -65,6 +65,19 @@ public final class ClientBalanceConsumerExample {
private static final ConcurrentHashMap<String, Long> partitionOffsetMap =
new ConcurrentHashMap<>();
+ /**
+ * Consume messages in Pull mode by assigning partitions by the client.
+ *
+ * @param args Startup parameter array, including the following parts:
+ * The 1st parameter masterServers is the master address(es)
to connect to,
+ * format is master1_ip:port[,master2_ip:port];
+ * The 2nd parameter subTopicAndFiterItems the topic(s) (and
filter condition set) to consume on,
+ * format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
+ * The 3rd parameter groupName is the name of consumer group;
+ * The 4th parameter consumeCount is the amount of messages
that need to be consumed;
+ * The 5th parameter totalGroupNodeCnt is the total number
of clients started by the consumer group;
+ * The 6th parameter fetchThreadCnt is the count of fetch
thread.
+ */
public static void main(String[] args) throws Throwable {
// 1. get and initial parameters
final String masterServers = args[0];
diff --git
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MAMessageProducerExample.java
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MAMessageProducerExample.java
index 6bccd9e..6a3b5de 100644
---
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MAMessageProducerExample.java
+++
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MAMessageProducerExample.java
@@ -54,14 +54,19 @@ public class MAMessageProducerExample {
private static final AtomicLong totalSentCnt = new AtomicLong(0);
private static ExecutorService sendExecutorService;
- // The 1st parameter masterServers format is
master1_ip:port[,master2_ip:port],
- // the master address(es) to connect to;
- // The 2nd parameter pubTopicAndFilterItems format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]]
- // the topic(s) (and filter condition set) to publish to.
- // The 3rd parameter msgCount is the message amount that needs to be sent
- // The 4th parameter pkgSize is the message's body size that needs to be
sent
- // The 5th parameter clientCount is the amount of producer
- // The 6th parameter sessionFactoryCnt is the amount of session factory
+ /**
+ * Produce messages through multi-session factory instances.
+ *
+ * @param args Startup parameter array, including the following parts:
+ * The 1st parameter masterServers is the master address(es)
to connect to,
+ * format is master1_ip:port[,master2_ip:port];
+ * The 2nd parameter pubTopicAndFilterItems is the topic(s)
(and filter condition set) to publish to,
+ * format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
+ * The 3rd parameter msgCount is the message amount that
needs to be sent;
+ * The 4th parameter pkgSize is the message's body size that
needs to be sent;
+ * The 5th parameter clientCount is the amount of producer;
+ * The 6th parameter sessionFactoryCnt is the amount of
session factory.
+ */
public static void main(String[] args) throws Throwable {
// 1. get call parameters
final String masterServers = args[0];
diff --git
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessageProducerExample.java
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessageProducerExample.java
index f87a068..6f62e6a 100644
---
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessageProducerExample.java
+++
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessageProducerExample.java
@@ -49,12 +49,17 @@ public final class MessageProducerExample {
private static MessageSessionFactory sessionFactory;
private static MessageProducer messageProducer;
- // The 1st parameter masterServers format is
master1_ip:port[,master2_ip:port],
- // the master address(es) to connect to;
- // The 2nd parameter pubTopicAndFilterItems format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]]
- // the topic(s) (and filter condition set) to publish to.
- // The 3rd parameter msgCount is the message amount that needs to be sent
- // The 4th parameter pkgSize is the message's body size that needs to be
sent
+ /**
+ * Produce messages through a single-session factory instance.
+ *
+ * @param args Startup parameter array, including the following parts:
+ * The 1st parameter masterServers is the master address(es)
to connect to,
+ * format is master1_ip:port[,master2_ip:port];
+ * The 2nd parameter pubTopicAndFilterItems is the topic(s)
(and filter condition set) to publish to,
+ * format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
+ * The 3rd parameter msgCount is the message amount that
needs to be sent;
+ * The 4th parameter pkgSize is the message's body size that
needs to be sent.
+ */
public static void main(String[] args) throws Throwable {
// 1. get and initial parameters
final String masterServers = args[0];
diff --git
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullConsumerExample.java
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullConsumerExample.java
index 393b3c4..4d32a2d 100644
---
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullConsumerExample.java
+++
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullConsumerExample.java
@@ -50,13 +50,18 @@ public final class MessagePullConsumerExample {
private static PullMessageConsumer pullConsumer;
private static MessageSessionFactory sessionFactory;
- // The 1st parameter masterServers format is
master1_ip:port[,master2_ip:port],
- // the master address(es) to connect to;
- // The 2nd parameter subTopicAndFiterItems format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]]
- // the topic(s) (and filter condition set) to consume on.
- // The 3rd parameter groupName is the name of consumer group
- // The 4th parameter consumeCount is the amount of messages that need to
be consumed
- // The 5th parameter fetchThreadCnt is the count of fetch thread
+ /**
+ * Consume messages in Pull mode through a single-session factory instance.
+ *
+ * @param args Startup parameter array, including the following parts:
+ * The 1st parameter masterServers is the master address(es)
to connect to,
+ * format is master1_ip:port[,master2_ip:port];
+ * The 2nd parameter subTopicAndFiterItems the topic(s) (and
filter condition set) to consume on,
+ * format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
+ * The 3rd parameter groupName is the name of consumer group;
+ * The 4th parameter consumeCount is the amount of messages
that need to be consumed;
+ * The 5th parameter fetchThreadCnt is the count of fetch
thread.
+ */
public static void main(String[] args) throws Throwable {
// 1. get and initial parameters
final String masterServers = args[0];
diff --git
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullSetConsumerExample.java
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullSetConsumerExample.java
index 827fd5d..47b441f 100644
---
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullSetConsumerExample.java
+++
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullSetConsumerExample.java
@@ -60,13 +60,18 @@ public final class MessagePullSetConsumerExample {
private static final ConcurrentHashMap<String, MessageSessionFactory>
multSessFtyMap =
new ConcurrentHashMap<>();
- // The 1st parameter masterServers format is
master1_ip:port[,master2_ip:port],
- // the master address(es) to connect to;
- // The 2nd parameter subTopicAndFiterItems format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]]
- // the topic(s) (and filter condition set) to consume on.
- // The 3rd parameter groupName is the name of consumer group
- // The 4th parameter consumeCount is the amount of messages that need to
be consumed
- // The 5th parameter fetchThreadCnt is the count of fetch thread
+ /**
+ * Consume messages with setting bootstrap offset in Pull mode through
multi-session factory instances.
+ *
+ * @param args Startup parameter array, including the following parts:
+ * The 1st parameter masterServers is the master address(es)
to connect to,
+ * format is master1_ip:port[,master2_ip:port];
+ * The 2nd parameter subTopicAndFiterItems the topic(s) (and
filter condition set) to consume on,
+ * format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
+ * The 3rd parameter groupName is the name of consumer group;
+ * The 4th parameter consumeCount is the amount of messages
that need to be consumed;
+ * The 5th parameter fetchThreadCnt is the count of fetch
thread.
+ */
public static void main(String[] args) throws Throwable {
// 1. get and initial parameters
final String masterServers = args[0];
diff --git
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePushConsumerExample.java
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePushConsumerExample.java
index 94c8d15..1ed9ef4 100644
---
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePushConsumerExample.java
+++
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePushConsumerExample.java
@@ -54,13 +54,18 @@ public final class MessagePushConsumerExample {
private static MessageSessionFactory sessionFactory;
private static final Map<String, PushMessageConsumer> consumerMap = new
HashMap<>();
- // The 1st parameter masterServers format is
master1_ip:port[,master2_ip:port],
- // the master address(es) to connect to;
- // The 2nd parameter subTopicAndFiterItems format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]]
- // the topic(s) (and filter condition set) to consume on.
- // The 3rd parameter groupName is the name of consumer group
- // The 4th parameter clientCount is the amount of consumer
- // The 5th parameter fetchThreadCnt is the count of fetch thread
+ /**
+ * Consume messages in Push mode through a single-session factory instance.
+ *
+ * @param args Startup parameter array, including the following parts:
+ * The 1st parameter masterServers is the master address(es)
to connect to,
+ * format is master1_ip:port[,master2_ip:port];
+ * The 2nd parameter subTopicAndFiterItems the topic(s) (and
filter condition set) to consume on,
+ * format is
topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
+ * The 3rd parameter groupName is the name of consumer group;
+ * The 4th parameter clientCount is the amount of consumer;
+ * The 5th parameter fetchThreadCnt is the count of fetch
thread.
+ */
public static void main(String[] args) throws Throwable {
// 1. get and initial parameters
final String masterServers = args[0];
diff --git
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MsgSendReceiveStats.java
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MsgSendReceiveStats.java
index cfc8a55..7a133fb 100644
---
a/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MsgSendReceiveStats.java
+++
b/inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MsgSendReceiveStats.java
@@ -56,10 +56,10 @@ public class MsgSendReceiveStats implements Runnable {
}
if (isProducer) {
logger.info("********* Current {} Message sent count
is {}, dlt is {}",
- new Object[]{entry.getKey(), currCount,
(currCount - befCount.get())});
+ entry.getKey(), currCount, (currCount -
befCount.get()));
} else {
logger.info("********* Current {} Message received
count is {}, dlt is {}",
- new Object[]{entry.getKey(), currCount,
(currCount - befCount.get())});
+ entry.getKey(), currCount, (currCount -
befCount.get()));
}
befCountMap.get(entry.getKey()).set(currCount);
}
@@ -70,6 +70,12 @@ public class MsgSendReceiveStats implements Runnable {
}
}
+ /**
+ * Record the number of messages by topicName dimension
+ *
+ * @param topicName topic name
+ * @param msgCnt message count
+ */
public void addMsgCount(final String topicName, int msgCnt) {
if (msgCnt > 0) {
AtomicLong currCount = counterMap.get(topicName);
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerConfig.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerConfig.java
index 8055695..4791149 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerConfig.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerConfig.java
@@ -34,7 +34,7 @@ import org.ini4j.Profile.Section;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Config of broker. Read from broker.ini config file.
*/
public class BrokerConfig extends AbstractFileConfig {
@@ -197,7 +197,7 @@ public class BrokerConfig extends AbstractFileConfig {
}
}
- /***
+ /**
* Load config from broker.ini by section.
*
* @param iniConf configure section
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerServiceServer.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerServiceServer.java
index 4a749c4..a30369a 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerServiceServer.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerServiceServer.java
@@ -85,7 +85,7 @@ import org.apache.inlong.tubemq.server.common.utils.RowLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Broker service. Receive and conduct client's request, store messages, query
messages, print statistics, etc.
*/
public class BrokerServiceServer implements BrokerReadService,
BrokerWriteService, Server {
@@ -137,7 +137,7 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
this.tubeConfig.getConsumerRegTimeoutMs(), consumerListener);
}
- /***
+ /**
* Start broker service
*
* @throws Exception
@@ -212,7 +212,7 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
this.started.set(true);
}
- /***
+ /**
* Stop broker service.
*/
@Override
@@ -226,26 +226,26 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
logger.info("BrokerService server stopped");
}
- /***
+ /**
* Get broker's registered consumer info
*
- * @return
+ * @return query result
*/
public Map<String, ConsumerNodeInfo> getConsumerRegisterMap() {
return consumerRegisterMap;
}
- /***
+ /**
* Get consumer's info by store key.
*
- * @param storeKey
- * @return
+ * @param storeKey the query key
+ * @return the consumer nodeinfo object
*/
public ConsumerNodeInfo getConsumerNodeInfo(String storeKey) {
return consumerRegisterMap.get(storeKey);
}
- /***
+ /**
* Get consumer's register time.
*
* @param consumerId
@@ -262,14 +262,14 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
return (timeoutInfo.getTimeoutTime() -
heartbeatManager.getConsumerTimeoutDlt());
}
- /***
+ /**
* Handle consumer's getMessageRequest.
*
- * @param request
- * @param rmtAddress
- * @param overtls
- * @return
- * @throws Throwable
+ * @param request the http request
+ * @param rmtAddress the remote node address
+ * @param overtls whether over TLS
+ * @return the response message
+ * @throws Throwable the exception during processing
*/
@Override
public GetMessageResponseB2C getMessagesC2B(GetMessageRequestC2B request,
@@ -416,23 +416,23 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
}
}
- /***
+ /**
* Query offset, then read data.
*
- * @param msgStore
- * @param consumerNodeInfo
- * @param group
- * @param topic
- * @param partitionId
- * @param lastConsumed
- * @param isManualCommitOffset
- * @param sentAddr
- * @param brokerAddr
- * @param rmtAddrInfo
- * @param isEscFlowCtrl
- * @param sb
- * @return
- * @throws IOException
+ * @param msgStore the message store instance
+ * @param consumerNodeInfo the consumer node instance
+ * @param group the consume group name
+ * @param topic the topic name
+ * @param partitionId the partition id
+ * @param lastConsumed whether the last messages has been
consumed
+ * @param isManualCommitOffset whether manual commit offset
+ * @param sentAddr the remote ip
+ * @param brokerAddr the broker ip
+ * @param rmtAddrInfo the remote address
+ * @param isEscFlowCtrl whether escape flow control
+ * @param sb the string buffer
+ * @return the query result
+ * @throws IOException the exception during processing
*/
private GetMessageResult getMessages(final MessageStore msgStore,
final ConsumerNodeInfo
consumerNodeInfo,
@@ -487,16 +487,16 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
}
}
- /***
+ /**
* Get message snapshot by given parameters.
*
- * @param topicName
- * @param partitionId
- * @param msgCount
- * @param filterCondSet
- * @param sb
- * @return
- * @throws Exception
+ * @param topicName the topic name
+ * @param partitionId the partition id
+ * @param msgCount the message count need to query
+ * @param filterCondSet the filter condition set
+ * @param sb the string buffer
+ * @return the messages returned
+ * @throws Exception the exception during processing
*/
public StringBuilder getMessageSnapshot(String topicName, int partitionId,
int msgCount, final Set<String>
filterCondSet,
@@ -553,7 +553,7 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
List<String> transferMessageList = new ArrayList<>();
List<TransferedMessage> tmpMsgList =
getMessageResult.transferedMessageList;
List<Message> messageList =
DataConverterUtil.convertMessage(topicName, tmpMsgList);
- int startPos = messageList.size() - msgCount < 0 ? 0 :
messageList.size() - msgCount;
+ int startPos = Math.max(messageList.size() - msgCount, 0);
for (; startPos < messageList.size(); startPos++) {
String msgItem = new String(
Base64.encodeBase64(messageList.get(startPos).getData()));
@@ -578,14 +578,14 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
}
}
- /***
+ /**
* Handle producer's sendMessage request.
*
- * @param request
- * @param rmtAddress
- * @param overtls
- * @return
- * @throws Throwable
+ * @param request the request
+ * @param rmtAddress the remote ip
+ * @param overtls whether transfer over TLS
+ * @return the response
+ * @throws Throwable the exception during processing
*/
@Override
public SendMessageResponseB2P sendMessageP2B(SendMessageRequestP2B request,
@@ -704,7 +704,7 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
}
}
- /***
+ /**
* append group current offset to storage
*
* @param groupOffsetMap group offset information
@@ -789,14 +789,14 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
}
}
- /***
+ /**
* Handle consumer register request.
*
- * @param request
- * @param rmtAddress
- * @param overtls
- * @return
- * @throws Throwable
+ * @param request the request
+ * @param rmtAddress the remote address
+ * @param overtls whether transfer over TLS
+ * @return the response
+ * @throws Throwable the exception during processing
*/
@Override
public RegisterResponseB2C consumerRegisterC2B(RegisterRequestC2B request,
@@ -899,19 +899,19 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
}
}
- /***
+ /**
* Handle consumer's register request.
*
- * @param clientId
- * @param groupName
- * @param topicName
- * @param partStr
- * @param filterCondSet
- * @param overtls
- * @param request
- * @param builder
- * @param strBuffer
- * @return
+ * @param clientId the client id
+ * @param groupName the group name
+ * @param topicName the topic name
+ * @param partStr the group-topic-partitionId key
+ * @param filterCondSet the filter condition set
+ * @param overtls whether transfer over TLS
+ * @param request the request
+ * @param builder the response builder
+ * @param strBuffer the string buffer
+ * @return the response
*/
private RegisterResponseB2C inProcessConsumerRegister(final String
clientId, final String groupName,
final String
topicName, final String partStr,
@@ -995,18 +995,18 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
}
}
- /***
+ /**
* Handle consumer's unregister request.
*
- * @param clientId
- * @param groupName
- * @param topicName
- * @param partStr
- * @param request
- * @param overtls
- * @param builder
- * @param strBuffer
- * @return
+ * @param clientId the client id
+ * @param groupName the group name
+ * @param topicName the topic name
+ * @param partStr the group-topic-partitionId key
+ * @param request the request
+ * @param overtls whether transfer over TLS
+ * @param builder the response builder
+ * @param strBuffer the string buffer
+ * @return the response
*/
private RegisterResponseB2C inProcessConsumerUnregister(final String
clientId, final String groupName,
final String
topicName, final String partStr,
@@ -1064,14 +1064,14 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
return builder.build();
}
- /***
+ /**
* Handle consumer's heartbeat request.
*
- * @param request
- * @param rmtAddress
- * @param overtls
- * @return
- * @throws Throwable
+ * @param request the request
+ * @param rmtAddress the remote address
+ * @param overtls whether transfer over TLS
+ * @return the response
+ * @throws Throwable the exception during processing
*/
@Override
public HeartBeatResponseB2C consumerHeartbeatC2B(HeartBeatRequestC2B
request,
@@ -1185,14 +1185,14 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
return builder.build();
}
- /***
+ /**
* Handle consumer's commit offset request.
*
- * @param request
- * @param rmtAddress
- * @param overtls
- * @return
- * @throws Throwable
+ * @param request the request
+ * @param rmtAddress the remote address
+ * @param overtls whether transfer over TLS
+ * @return the response
+ * @throws Throwable the exception during processing
*/
@Override
public CommitOffsetResponseB2C consumerCommitC2B(CommitOffsetRequestC2B
request,
@@ -1310,7 +1310,7 @@ public class BrokerServiceServer implements
BrokerReadService, BrokerWriteServic
: consumerNodeInfo.getQryPriorityId();
}
- /***
+ /**
* Consumer timeout handler. Update consumer's info if exists consumer
timeout.
*/
public class ConsumerTimeoutListener implements TimeoutListener {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java
index f6f52cf..1f57fac 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java
@@ -17,6 +17,7 @@
package org.apache.inlong.tubemq.server.broker;
+import java.io.IOException;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -61,7 +62,7 @@ import org.apache.log4j.LogManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Tube broker server. In charge of init each components, and communicating to
master.
*/
public class TubeBroker implements Stoppable {
@@ -96,13 +97,19 @@ public class TubeBroker implements Stoppable {
private MasterService masterService;
private boolean requireReportConf = false;
private boolean isOnline = false;
- private AtomicBoolean shutdown = new AtomicBoolean(true);
+ private final AtomicBoolean shutdown = new AtomicBoolean(true);
private final AtomicBoolean isKeepAlive = new AtomicBoolean(false);
private final AtomicLong lastRegTime = new AtomicLong(0);
- private AtomicBoolean shutdownHooked = new AtomicBoolean(false);
- private AtomicLong heartbeatErrors = new AtomicLong(0);
+ private final AtomicBoolean shutdownHooked = new AtomicBoolean(false);
+ private final AtomicLong heartbeatErrors = new AtomicLong(0);
private int maxReleaseTryCnt = 10;
+ /**
+ * Initial broker instance.
+ *
+ * @param tubeConfig the initial configure
+ * @throws IOException the exception during processing
+ */
public TubeBroker(final BrokerConfig tubeConfig) throws Exception {
java.security.Security.setProperty("networkaddress.cache.ttl", "3");
java.security.Security.setProperty("networkaddress.cache.negative.ttl", "1");
@@ -191,10 +198,10 @@ public class TubeBroker implements Stoppable {
return brokerServiceServer;
}
- /***
+ /**
* Start broker service.
*
- * @throws Exception
+ * @throws Exception the exception during processing
*/
public void start() throws Exception {
logger.info("Starting tube server...");
@@ -393,10 +400,10 @@ public class TubeBroker implements Stoppable {
}
}
- /***
+ /**
* Register to master. Try multi times if failed.
*
- * @throws StartupException
+ * @throws StartupException the exception during processing
*/
private void register2Master() throws StartupException {
int remainingRetry = 5;
@@ -502,11 +509,11 @@ public class TubeBroker implements Stoppable {
return defSetting;
}
- /***
+ /**
* Build register request to master.
*
- * @return
- * @throws Exception
+ * @return the register request object
+ * @throws Exception the exception during processing
*/
private RegisterRequestB2M createMasterRegisterRequest() throws Exception {
RegisterRequestB2M.Builder builder = RegisterRequestB2M.newBuilder();
@@ -550,10 +557,10 @@ public class TubeBroker implements Stoppable {
return builder.build();
}
- /***
+ /**
* Build heartbeat request to master.
*
- * @return
+ * @return the HeartRequestB2M request object
*/
private HeartRequestB2M createBrokerHeartBeatRequest() {
HeartRequestB2M.Builder builder = HeartRequestB2M.newBuilder();
@@ -602,10 +609,10 @@ public class TubeBroker implements Stoppable {
return builder.build();
}
- /***
+ /**
* Build close request to master.
*
- * @return
+ * @return the CloseRequestB2M request object
*/
private CloseRequestB2M createMasterCloseRequest() {
CloseRequestB2M.Builder builder = CloseRequestB2M.newBuilder();
@@ -617,10 +624,10 @@ public class TubeBroker implements Stoppable {
return builder.build();
}
- /***
+ /**
* Build master certificate info.
*
- * @return
+ * @return the MasterCertificateInfo builder
*/
private ClientMaster.MasterCertificateInfo.Builder
genMasterCertificateInfo() {
ClientMaster.MasterCertificateInfo.Builder authInfoBuilder = null;
@@ -633,7 +640,7 @@ public class TubeBroker implements Stoppable {
return authInfoBuilder;
}
- /***
+ /**
* Shutdown hook.
*/
private final class ShutdownHook extends Thread {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/exception/OffsetStoreException.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/exception/OffsetStoreException.java
index 638b174..9bd2f44 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/exception/OffsetStoreException.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/exception/OffsetStoreException.java
@@ -17,7 +17,7 @@
package org.apache.inlong.tubemq.server.broker.exception;
-/***
+/**
* Save offset occur error throw this Exception.
*/
public class OffsetStoreException extends Exception {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/exception/StartupException.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/exception/StartupException.java
index b7e4cea..92081cc 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/exception/StartupException.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/exception/StartupException.java
@@ -17,7 +17,7 @@
package org.apache.inlong.tubemq.server.broker.exception;
-/***
+/**
* Service startup occur error throw this Exception. eg: startup config
parameters illegal error.
*/
public class StartupException extends RuntimeException {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerDefMetadata.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerDefMetadata.java
index 3b85fb1..95e4430 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerDefMetadata.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerDefMetadata.java
@@ -20,7 +20,7 @@ package org.apache.inlong.tubemq.server.broker.metadata;
import org.apache.inlong.tubemq.corebase.TokenConstants;
import org.apache.inlong.tubemq.corebase.utils.TStringUtils;
-/***
+/**
* Default metadata for broker, it mainly contains topic default
config(partitions count, delete policy...).
* These metadata will be overwrite if explicitly be set.
*/
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerMetadataManager.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerMetadataManager.java
index ce8e033..47c0ac0 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerMetadataManager.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/BrokerMetadataManager.java
@@ -31,7 +31,7 @@ import
org.apache.inlong.tubemq.server.common.TStatusConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Broker's metadata management util. Metadata contains broker's default
configurations, topics,
* topic that will be deleted, and broker's policy definition.
* Metadata is got from Master service, it will refresh in heartbeat between
broker and master.
@@ -168,7 +168,7 @@ public class BrokerMetadataManager implements
MetadataManager {
return removedTopicConfigMap;
}
- /***
+ /**
* Get hard removed topics. Hard removed means the disk files is deleted,
cannot be recovery.
* Topic will be deleted in two phases, the first is mark topic's file
delete, the second is delete the disk files.
*
@@ -204,7 +204,7 @@ public class BrokerMetadataManager implements
MetadataManager {
this.lastRptBrokerMetaConfId = lastRptBrokerMetaConfId;
}
- /***
+ /**
* Update broker's metadata in memory, then fire these metadata take
effect.
* These params are got from Master Service.
*
@@ -276,7 +276,7 @@ public class BrokerMetadataManager implements
MetadataManager {
this.propertyChangeSupport.firePropertyChange("unflushInterval", null,
null);
}
- /***
+ /**
* Update will be deleted topics info. These params are got from Master
Service.
*
* @param isTakeRemoveTopics whether take removed topics
@@ -345,7 +345,7 @@ public class BrokerMetadataManager implements
MetadataManager {
this.propertyChangeSupport.addPropertyChangeListener(propertyName,
listener);
}
- /***
+ /**
* Add historical offset storage topic by default
*
* @param brokerDefMeta broker default meta configure
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/MetadataManager.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/MetadataManager.java
index e03d2c5..0fbb226 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/MetadataManager.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/MetadataManager.java
@@ -22,7 +22,7 @@ import java.util.List;
import java.util.Map;
import org.apache.inlong.tubemq.corebase.policies.FlowCtrlRuleHandler;
-/***
+/**
* Metadata's management interface.
*/
public interface MetadataManager {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/TopicMetadata.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/TopicMetadata.java
index a0e4d09..77116a3 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/TopicMetadata.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/metadata/TopicMetadata.java
@@ -28,7 +28,7 @@ import org.apache.inlong.tubemq.corebase.utils.TStringUtils;
import org.apache.inlong.tubemq.corebase.utils.Tuple2;
import org.apache.inlong.tubemq.server.common.TStatusConstants;
-/***
+/**
* Topic's metadata. Contains topic name, partitions count, etc.
*/
public class TopicMetadata {
@@ -67,7 +67,7 @@ public class TopicMetadata {
// the allowed min memory cache size
private int minMemCacheSize = TBaseConstants.META_VALUE_UNDEFINED;
- /***
+ /**
* Build TopicMetadata from brokerDefMetadata(default config) and
topicMetaConfInfo(custom config).
*
* @param brokerDefMetadata the default topic meta configure
@@ -158,7 +158,7 @@ public class TopicMetadata {
}
}
- /***
+ /**
* Build TopicMetadata by default topic meta and the special field values.
*
* @param brokerDefMetadata the default topic meta configure
@@ -468,7 +468,7 @@ public class TopicMetadata {
return true;
}
- /***
+ /**
* Each property will be compared, in case of the new added properties.
*
* @param other the compare object
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStore.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStore.java
index d27e78e..3764559 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStore.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStore.java
@@ -54,7 +54,7 @@ import org.apache.inlong.tubemq.server.common.utils.IdWorker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Topic's message storage. It's a logical topic storage. Contains multi types
storage: data in memory,
* data in disk, and statistics of produce and consume.
*/
@@ -151,7 +151,7 @@ public class MessageStore implements Closeable {
this.lastMemFlushTime.set(System.currentTimeMillis());
}
- /***
+ /**
* Get message from message store. Support the given offset, filter.
*
* @param reqSwitch read message from where
@@ -162,7 +162,7 @@ public class MessageStore implements Closeable {
* @param msgSizeLimit the max read size
* @param reqRcvTime the timestamp of the record to be checked
* @return read result
- * @throws IOException exception while process
+ * @throws IOException the exception during processing
*/
public GetMessageResult getMessages(int reqSwitch, long requestOffset,
int partitionId, ConsumerNodeInfo
consumerNodeInfo,
@@ -298,7 +298,7 @@ public class MessageStore implements Closeable {
return retResult;
}
- /***
+ /**
* Get start offset by timestamp.
*
* @param timestamp timestamp
@@ -326,7 +326,7 @@ public class MessageStore implements Closeable {
}
}
- /***
+ /**
* Append msg to store.
*
* @param appendResult the append result
@@ -339,7 +339,7 @@ public class MessageStore implements Closeable {
* @param sentAddr the address to send the message to
*
* @return the process result
- * @throws IOException exception while process
+ * @throws IOException the exception during processing
*/
public boolean appendMsg(AppendResult appendResult, int dataLength,
int dataCheckSum, byte[] data,
@@ -350,7 +350,7 @@ public class MessageStore implements Closeable {
System.currentTimeMillis(), 3, 2);
}
- /***
+ /**
* Append msg to store.
*
* @param appendResult the append result
@@ -366,7 +366,7 @@ public class MessageStore implements Closeable {
* @param waitRetryMs the wait duration while retry
*
* @return the process result
- * @throws IOException exception while process
+ * @throws IOException the exception during processing
*/
public boolean appendMsg2(AppendResult appendResult, int dataLength,
int dataCheckSum, byte[] data,
@@ -429,7 +429,7 @@ public class MessageStore implements Closeable {
return this.msgFileStatisInfo;
}
- /***
+ /**
* Execute cleanup policy.
*
* @param onlyCheck whether only check status
@@ -445,7 +445,7 @@ public class MessageStore implements Closeable {
return msgFileStore.runClearupPolicy(onlyCheck);
}
- /***
+ /**
* Refresh unflush threshold
*
* @param topicMetadata topic meta data
@@ -481,10 +481,10 @@ public class MessageStore implements Closeable {
}
}
- /***
+ /**
* Flush file store to disk.
*
- * @throws IOException exception while process
+ * @throws IOException the exception during processing
*/
public void flushFile() throws IOException {
if (this.closed.get()) {
@@ -495,10 +495,10 @@ public class MessageStore implements Closeable {
msgFileStore.flushDiskFile();
}
- /***
+ /**
* Flush memory store to file.
*
- * @throws IOException exception while process
+ * @throws IOException the exception during processing
*/
public void flushMemCacheData() throws IOException {
if (this.closed.get()) {
@@ -669,7 +669,7 @@ public class MessageStore implements Closeable {
return memCacheSize;
}
- /***
+ /**
* Append message and trigger flush operation.
*
* @param partitionId the partitionId for reading messages
@@ -682,7 +682,7 @@ public class MessageStore implements Closeable {
* @param appendResult the append result
*
* @return the append result
- * @throws IOException exception while process
+ * @throws IOException the exception during processing
*/
private boolean triggerFlushAndAddMsg(int partitionId, int keyCode,
long receivedTime, int entryLength,
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStoreManager.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStoreManager.java
index 040927f..6f10324 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStoreManager.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStoreManager.java
@@ -59,7 +59,7 @@ import
org.apache.inlong.tubemq.server.common.TStatusConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Message storage management. It contains all topics on broker. In charge of
store, expire, and flush operation,
*/
public class MessageStoreManager implements StoreService {
@@ -85,6 +85,13 @@ public class MessageStoreManager implements StoreService {
// the status that is deleting topic.
private final AtomicBoolean isRemovingTopic = new AtomicBoolean(false);
+ /**
+ * Initial the message-store manager.
+ *
+ * @param tubeBroker the broker instance
+ * @param tubeConfig the initial configure
+ * @throws IOException the exception during processing
+ */
public MessageStoreManager(final TubeBroker tubeBroker,
final BrokerConfig tubeConfig) throws
IOException {
super();
@@ -261,7 +268,7 @@ public class MessageStoreManager implements StoreService {
}
}
- /***
+ /**
* Get message store by topic.
*
* @param topic query topic name
@@ -277,13 +284,13 @@ public class MessageStoreManager implements StoreService {
return map.values();
}
- /***
+ /**
* Get or create message store.
*
- * @param topic
- * @param partition
- * @return
- * @throws IOException
+ * @param topic the topic name
+ * @param partition the partition id
+ * @return the message-store instance
+ * @throws IOException the exception during processing
*/
@Override
public MessageStore getOrCreateMessageStore(final String topic,
@@ -339,16 +346,16 @@ public class MessageStoreManager implements StoreService {
return this.tubeBroker;
}
- /***
+ /**
* Get message from store.
*
- * @param msgStore
- * @param topic
- * @param partitionId
- * @param msgCount
- * @param filterCondSet
- * @return
- * @throws IOException
+ * @param msgStore the message-store
+ * @param topic the topic name
+ * @param partitionId the partition id
+ * @param msgCount the message count to read
+ * @param filterCondSet the filter condition set
+ * @return the query result
+ * @throws IOException the exception during processing
*/
public GetMessageResult getMessages(final MessageStore msgStore,
final String topic,
@@ -387,7 +394,7 @@ public class MessageStoreManager implements StoreService {
return Collections.unmodifiableMap(this.dataStores);
}
- /***
+ /**
* Query topic's publish info.
*
* @param topicSet query's topic set
@@ -444,7 +451,7 @@ public class MessageStoreManager implements StoreService {
return topicPubStoreInfoMap;
}
- /***
+ /**
* Query topic's publish info.
*
* @param groupOffsetMap query's topic set
@@ -517,12 +524,12 @@ public class MessageStoreManager implements StoreService {
return fileSet;
}
- /***
+ /**
* Load stores sequential.
*
- * @param tubeConfig
- * @throws IOException
- * @throws InterruptedException
+ * @param tubeConfig the broker's configure
+ * @throws IOException the exception during processing
+ * @throws InterruptedException the exception during processing
*/
private void loadMessageStores(final BrokerConfig tubeConfig)
throws IOException, InterruptedException {
@@ -620,11 +627,11 @@ public class MessageStoreManager implements StoreService {
.append((System.currentTimeMillis() - start) / 1000).append("
secs").toString());
}
- /***
+ /**
* Load stores in parallel.
*
- * @param tasks
- * @throws InterruptedException
+ * @param tasks the load tasks
+ * @throws InterruptedException the exception during processing
*/
private void loadStoresInParallel(List<Callable<MessageStore>> tasks)
throws InterruptedException {
ExecutorService executor =
@@ -661,6 +668,12 @@ public class MessageStoreManager implements StoreService {
}
}
+ /**
+ * Refresh message-store's dynamic configures
+ *
+ * @param oldTopicConfigMap the stored topic configure map
+ * @param newTopicConfigMap the newly topic configure map
+ */
public void refreshMessageStoresHoldVals(Map<String, TopicMetadata>
oldTopicConfigMap,
Map<String, TopicMetadata>
newTopicConfigMap) {
if (((newTopicConfigMap == null) || newTopicConfigMap.isEmpty())
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/StoreService.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/StoreService.java
index f339f3e..c34702c 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/StoreService.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/StoreService.java
@@ -24,7 +24,7 @@ import java.util.Set;
import org.apache.inlong.tubemq.server.broker.offset.OffsetRecordInfo;
import org.apache.inlong.tubemq.server.broker.utils.TopicPubStoreInfo;
-/***
+/**
* Store service interface.
*/
public interface StoreService {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegment.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegment.java
index 6839a2a..bc357b2 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegment.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegment.java
@@ -32,7 +32,7 @@ import
org.apache.inlong.tubemq.server.broker.utils.DataStoreUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Segment file. Topic contains multi FileSegments. Each FileSegment contains
data file and index file.
* It is mini particle of topic expire policy. It will be marked deleted when
expired.
*/
@@ -199,7 +199,7 @@ public class FileSegment implements Segment {
}
}
- /***
+ /**
* Messages can only be appended to the last FileSegment.
* The last FileSegment is writable, the others are mutable.
*
@@ -237,7 +237,7 @@ public class FileSegment implements Segment {
return this.start + offset;
}
- /***
+ /**
* Flush file cache to disk.
*
* @param force whether to brush
@@ -276,7 +276,7 @@ public class FileSegment implements Segment {
&& offset <= this.start + this.getCachedSize() - 1);
}
- /***
+ /**
* Release reference to this FileSegment.
* File's channel will be closed when the reference decreased to 0.
*/
@@ -295,7 +295,7 @@ public class FileSegment implements Segment {
return start + cachedSize.get();
}
- /***
+ /**
* Return the position that have been flushed to disk.
*
* @return the position that have been flushed to disk
@@ -310,7 +310,7 @@ public class FileSegment implements Segment {
return mutable;
}
- /***
+ /**
* Set FileSegment to readonly.
*
* @param mutable mutable or immutable
@@ -390,7 +390,7 @@ public class FileSegment implements Segment {
}
}
- /***
+ /**
* read index record's append time.
* @param reqOffset request offset.
* @return message append time.
@@ -410,7 +410,7 @@ public class FileSegment implements Segment {
return readUnit.getLong(DataStoreUtils.INDEX_POS_TIME_RECV);
}
- /***
+ /**
* Check whether this FileSegment is expired, and set expire status.
* The last FileSegment cannot be marked expired.
*
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentList.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentList.java
index 09a0763..4a74a7d 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentList.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentList.java
@@ -22,7 +22,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* FileSegments management. Contains two types FileSegment: data and index.
*/
public class FileSegmentList implements SegmentList {
@@ -56,7 +56,7 @@ public class FileSegmentList implements SegmentList {
return segmentList.get();
}
- /***
+ /**
* Return segment by the given offset.
*
* @param offset the position to search
@@ -85,7 +85,7 @@ public class FileSegmentList implements SegmentList {
}
}
- /***
+ /**
* Check each FileSegment whether is expired, and set expire status.
*
* @param checkTimestamp current check timestamp
@@ -107,7 +107,7 @@ public class FileSegmentList implements SegmentList {
return hasExpired;
}
- /***
+ /**
* Check FileSegments whether is expired, close all expired FileSegments,
and then delete these files.
*
* @param sb string buffer
@@ -174,7 +174,7 @@ public class FileSegmentList implements SegmentList {
return curViews[curViews.length - 1];
}
- /***
+ /**
* Return the start position of these FileSegments.
*
* @return the first position
@@ -199,7 +199,7 @@ public class FileSegmentList implements SegmentList {
return last;
}
- /***
+ /**
* Return the max position of these FileSegments.
*
* @return the latest position
@@ -230,7 +230,7 @@ public class FileSegmentList implements SegmentList {
return last.getRightAppendTime();
}
- /***
+ /**
* Return the max position that have been flushed to disk.
*
* @return the latest committed offset
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/GetMessageResult.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/GetMessageResult.java
index e1002c5..5cd6fef 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/GetMessageResult.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/GetMessageResult.java
@@ -24,7 +24,7 @@ import org.apache.inlong.tubemq.corebase.TBaseConstants;
import
org.apache.inlong.tubemq.corebase.protobuf.generated.ClientBroker.TransferedMessage;
import org.apache.inlong.tubemq.server.broker.stats.CountItem;
-/***
+/**
* Broker's reply to Consumer's GetMessage request.
*/
public class GetMessageResult {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStatisInfo.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStatisInfo.java
index 3d1c427..49266f5 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStatisInfo.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStatisInfo.java
@@ -25,7 +25,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.inlong.tubemq.server.common.TServerConstants;
-/***
+/**
* Statistics of message file. Contains read, write, failed, etc.
*/
public class MsgFileStatisInfo {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStore.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStore.java
index 2ea3f49..6f7f568 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStore.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStore.java
@@ -22,7 +22,6 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -45,7 +44,7 @@ import org.apache.inlong.tubemq.server.common.utils.FileUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Message file's storage. Contains data file and index file.
*/
public class MsgFileStore implements Closeable {
@@ -103,7 +102,7 @@ public class MsgFileStore implements Closeable {
this.lastFlushTime.set(System.currentTimeMillis());
}
- /***
+ /**
* Batch append message to file segment
*
* @param sb string buffer
@@ -222,7 +221,7 @@ public class MsgFileStore implements Closeable {
}
}
- /***
+ /**
* Get message from index and data files.
*
* @param partitionId the partitionId for reading messages
@@ -383,7 +382,7 @@ public class MsgFileStore implements Closeable {
totalSize, countMap, transferedMessageList);
}
- /***
+ /**
* Get the segment start Offset that contains the specified timestamp
*
* @param timestamp the specified timestamp
@@ -455,7 +454,7 @@ public class MsgFileStore implements Closeable {
}
}
- /***
+ /**
* Clean expired data files and index files.
*
* @param onlyCheck whether to check only
@@ -480,10 +479,10 @@ public class MsgFileStore implements Closeable {
return (hasExpiredDataSegs || hasExpiredIndexSegs);
}
- /***
+ /**
* Flush data to disk at interval.
*
- * @throws IOException exception while process
+ * @throws IOException the exception during processing
*/
public void flushDiskFile() throws IOException {
long checkTimestamp = System.currentTimeMillis();
@@ -598,16 +597,10 @@ public class MsgFileStore implements Closeable {
accum.add(new FileSegment(offsetIfCreate, newFile, segType));
} else {
// The list of segments is required to be arranged continuously
from low to high
- Collections.sort(accum, new Comparator<Segment>() {
+ accum.sort(new Comparator<Segment>() {
@Override
public int compare(final Segment o1, final Segment o2) {
- if (o1.getStart() == o2.getStart()) {
- return 0;
- } else if (o1.getStart() > o2.getStart()) {
- return 1;
- } else {
- return -1;
- }
+ return Long.compare(o1.getStart(), o2.getStart());
}
});
validateSegments(segTypeStr, accum);
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/Segment.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/Segment.java
index 6ecd417..4a5fc51 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/Segment.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/Segment.java
@@ -21,14 +21,14 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
-/***
+/**
* Storage segment, usually implemented in file format.
*/
public interface Segment {
void close();
- /***
+ /**
* Messages can only be appended to the last FileSegment.
* The last FileSegment is writable, the others are mutable.
*
@@ -72,7 +72,7 @@ public interface Segment {
void relViewRef();
- /***
+ /**
* Read data to buffer from absolute position.
*
* @param bf buffer to store data
@@ -80,7 +80,7 @@ public interface Segment {
*/
void read(ByteBuffer bf, long absOffset) throws IOException;
- /***
+ /**
* Read data to buffer from relative position.
*
* @param bf buffer to store data
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/SegmentList.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/SegmentList.java
index cf4bed4..a5017ea 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/SegmentList.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/SegmentList.java
@@ -19,7 +19,7 @@ package org.apache.inlong.tubemq.server.broker.msgstore.disk;
import java.io.IOException;
-/***
+/**
* Segment list.
*/
public interface SegmentList {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/SegmentType.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/SegmentType.java
index 63bc83c..cb9d2b3 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/SegmentType.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/SegmentType.java
@@ -17,7 +17,7 @@
package org.apache.inlong.tubemq.server.broker.msgstore.disk;
-/***
+/**
* Segment's type, data and index.
*/
public enum SegmentType {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/GetCacheMsgResult.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/GetCacheMsgResult.java
index 876358b..da24016 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/GetCacheMsgResult.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/GetCacheMsgResult.java
@@ -20,7 +20,7 @@ package org.apache.inlong.tubemq.server.broker.msgstore.mem;
import java.nio.ByteBuffer;
import java.util.List;
-/***
+/**
* Result set of read operation from memory.
*/
public class GetCacheMsgResult {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStatisInfo.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStatisInfo.java
index 57762ce..6edf641 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStatisInfo.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStatisInfo.java
@@ -25,7 +25,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.inlong.tubemq.server.common.TServerConstants;
-/***
+/**
* Statistics of message memory. Contains read, write, failed, etc. It's
similar to MsgFileStatisInfo.
*/
public class MsgMemStatisInfo {
@@ -56,7 +56,7 @@ public class MsgMemStatisInfo {
}
}
- /***
+ /**
* Add write fail count statistic.
*/
public void addWriteFailCount() {
@@ -76,7 +76,7 @@ public class MsgMemStatisInfo {
}
}
- /***
+ /**
* Add full type count statistic.
*
* @param timeRecv
@@ -110,7 +110,7 @@ public class MsgMemStatisInfo {
}
}
- /***
+ /**
* Add memory flush count statistic.
*
* @param isTimeOut
@@ -135,10 +135,10 @@ public class MsgMemStatisInfo {
}
}
- /***
+ /**
* Add flush time statistic.
*
- * @param flushTIme
+ * @param flushTIme the flush time
*/
public void addFlushTimeStatis(long flushTIme) {
if (isStart.get()) {
@@ -159,7 +159,7 @@ public class MsgMemStatisInfo {
}
}
- /***
+ /**
* Add message size statistic.
*
* @param timeRecv
@@ -188,11 +188,11 @@ public class MsgMemStatisInfo {
}
}
- /***
+ /**
* Get current message size status info.
*
- * @param needRefresh
- * @return
+ * @param needRefresh Whether the statistics block needs to be refreshed
+ * @return the statistics data
*/
public String getCurMsgSizeStatisInfo(boolean needRefresh) {
MemStatisItemSet oldCountSet;
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStore.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStore.java
index d5c9c15..bad8da7 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStore.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStore.java
@@ -39,7 +39,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sun.nio.ch.DirectBuffer;
-/***
+/**
* Message's memory storage. It use direct memory store messages that received
but not have been flushed to disk.
*/
public class MsgMemStore implements Closeable {
@@ -84,7 +84,7 @@ public class MsgMemStore implements Closeable {
this.writeIndexStartPos = writeIndexStartPos;
}
- /***
+ /**
* Append message to memory cache
*
* @param msgMemStatisInfo statistical information object
@@ -145,7 +145,7 @@ public class MsgMemStore implements Closeable {
return true;
}
- /***
+ /**
* Read from memory, read index, then data.
*
* @param lstRdDataOffset the recent data offset read before
@@ -284,12 +284,12 @@ public class MsgMemStore implements Closeable {
lstRdIndexOffset, readedSize, lastDataRdOff, totalReadSize,
cacheMsgList);
}
- /***
+ /**
* Batch flush memory data to disk.
*
* @param msgFileStore the file storage
* @param strBuffer the message buffer
- * @throws IOException the exception while process
+ * @throws IOException the exception during processing
*/
public void batchFlush(MsgFileStore msgFileStore,
StringBuilder strBuffer) throws Throwable {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/nodeinfo/ConsumerNodeInfo.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/nodeinfo/ConsumerNodeInfo.java
index 80bce9d..3444daf 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/nodeinfo/ConsumerNodeInfo.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/nodeinfo/ConsumerNodeInfo.java
@@ -26,7 +26,7 @@ import
org.apache.inlong.tubemq.corebase.policies.FlowCtrlRuleHandler;
import org.apache.inlong.tubemq.server.broker.msgstore.MessageStoreManager;
import org.apache.inlong.tubemq.server.common.TServerConstants;
-/***
+/**
* Consumer node info, which broker contains.
*/
public class ConsumerNodeInfo {
@@ -36,14 +36,14 @@ public class ConsumerNodeInfo {
private final MessageStoreManager storeManager;
// consumer id
private String consumerId;
- private String sessionKey;
- private long sessionTime;
+ private final String sessionKey;
+ private final long sessionTime;
// is filter consumer or not
private boolean isFilterConsume = false;
// filter conditions in string format
- private Set<String> filterCondStrs = new HashSet<>(10);
+ private final Set<String> filterCondStrs = new HashSet<>(10);
// filter conditions in int format
- private Set<Integer> filterCondCode = new HashSet<>(10);
+ private final Set<Integer> filterCondCode = new HashSet<>(10);
// consumer's address
private String rmtAddrInfo;
private boolean isSupportLimit = false;
@@ -57,7 +57,7 @@ public class ConsumerNodeInfo {
private FlowCtrlResult curFlowCtrlVal =
new FlowCtrlResult(Long.MAX_VALUE, 0);
private long nextLimitUpdateTime = 0;
- private AtomicInteger qryPriorityId =
+ private final AtomicInteger qryPriorityId =
new AtomicInteger(TBaseConstants.META_VALUE_UNDEFINED);
private long createTime = System.currentTimeMillis();
@@ -226,7 +226,7 @@ public class ConsumerNodeInfo {
return this.rmtAddrInfo;
}
- /***
+ /**
* Recalculate message limit value.
*
* @param curDataDlt
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/DefaultOffsetManager.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/DefaultOffsetManager.java
index 83892b6..ad4a073 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/DefaultOffsetManager.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/DefaultOffsetManager.java
@@ -40,7 +40,7 @@ import
org.apache.inlong.tubemq.server.common.offsetstorage.ZkOffsetStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Default offset manager.
* Conduct consumer's commit offset operation and consumer's offset that has
consumed but not committed.
*/
@@ -91,7 +91,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
logger.info("[Offset Manager] Offset Manager service stopped!");
}
- /***
+ /**
* Load offset.
*
* @param msgStore
@@ -152,7 +152,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return regInfo;
}
- /***
+ /**
* Get offset by parameters.
*
* @param msgStore
@@ -236,7 +236,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
}
}
- /***
+ /**
* Commit offset.
*
* @param group
@@ -270,7 +270,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return updatedOffset;
}
- /***
+ /**
* Reset offset.
*
* @param store
@@ -305,7 +305,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return oldOffset;
}
- /***
+ /**
* Get temp offset.
*
* @param group
@@ -328,7 +328,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return 0;
}
- /***
+ /**
* Get in-memory and in zk group set
*
* @return booked group in memory and in zk
@@ -343,7 +343,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return groupSet;
}
- /***
+ /**
* Get in-memory group set
*
* @return booked group in memory
@@ -354,7 +354,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return cacheGroup;
}
- /***
+ /**
* Get in-zookeeper but not in memory's group set
*
* @return booked group in zookeeper
@@ -372,10 +372,11 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return unUsedGroups;
}
- /***
+ /**
* Get the topic set subscribed by the consumer group
- * @param group
- * @return topic set subscribed
+ *
+ * @param group the queries group name
+ * @return the topic set subscribed
*/
@Override
public Set<String> getGroupSubInfo(String group) {
@@ -395,7 +396,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return result;
}
- /***
+ /**
* Get group's offset by Specified topic-partitions
* @param group
* @param topicPartMap
@@ -453,7 +454,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return result;
}
- /***
+ /**
* Get online groups' offset information
*
* @return group offset info in memory or zk
@@ -488,7 +489,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return result;
}
- /***
+ /**
* Reset offset.
*
* @param groups the groups to reset offset
@@ -533,7 +534,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
return changed;
}
- /***
+ /**
* Delete offset.
*
* @param onlyMemory
@@ -580,7 +581,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
}
}
- /***
+ /**
* Set temp offset.
*
* @param group
@@ -623,7 +624,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
}
}
- /***
+ /**
* Commit temp offsets.
*/
private void commitTmpOffsets() {
@@ -660,7 +661,7 @@ public class DefaultOffsetManager extends
AbstractDaemonService implements Offse
BrokerMetricsHolder.updSyncZKDurations(System.currentTimeMillis() -
startTime);
}
- /***
+ /**
* Load or create offset.
*
* @param group
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetRecordInfo.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetRecordInfo.java
index a5f8b9b..d552f40 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetRecordInfo.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetRecordInfo.java
@@ -22,7 +22,7 @@ import java.util.Map;
import org.apache.inlong.tubemq.corebase.TokenConstants;
import org.apache.inlong.tubemq.corebase.utils.DateTimeConvertUtils;
-/***
+/**
* The offset snapshot of the consumer group on the broker.
*/
public class OffsetRecordInfo {
@@ -35,7 +35,7 @@ public class OffsetRecordInfo {
this.groupName = groupName;
}
- /***
+ /**
* Add confirmed offset of topic-partitionId.
*
* @param topicName topic name
@@ -58,7 +58,7 @@ public class OffsetRecordInfo {
return histOffsetMap;
}
- /***
+ /**
* Build consumption offset information in string format
*
* @param strBuff string buffer
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetRecordService.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetRecordService.java
index 90af50d..751bbe9 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetRecordService.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetRecordService.java
@@ -25,7 +25,7 @@ import
org.apache.inlong.tubemq.server.broker.msgstore.MessageStoreManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* History offset service
*
*/
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetService.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetService.java
index 842f2ac..5c68a31 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetService.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/OffsetService.java
@@ -26,7 +26,7 @@ import org.apache.inlong.tubemq.corebase.utils.Tuple3;
import org.apache.inlong.tubemq.server.broker.msgstore.MessageStore;
import org.apache.inlong.tubemq.server.common.offsetstorage.OffsetStorageInfo;
-/***
+/**
* Offset manager service interface.
*/
public interface OffsetService {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/RecordItem.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/RecordItem.java
index 6602d8e..ffe9da9 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/RecordItem.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/RecordItem.java
@@ -19,7 +19,7 @@ package org.apache.inlong.tubemq.server.broker.offset;
import org.apache.inlong.tubemq.corebase.TBaseConstants;
-/***
+/**
* The offset snapshot of the consumer group on the partition.
*/
public class RecordItem {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/CountItem.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/CountItem.java
index 03fa12a..58ed92f 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/CountItem.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/CountItem.java
@@ -19,7 +19,7 @@ package org.apache.inlong.tubemq.server.broker.stats;
import java.util.concurrent.atomic.AtomicLong;
-/***
+/**
* Statistic of message, contains message's count and message's size.
*/
public class CountItem {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/GroupCountService.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/GroupCountService.java
index b100ca1..a162dd3 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/GroupCountService.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/GroupCountService.java
@@ -26,7 +26,7 @@ import
org.apache.inlong.tubemq.corebase.daemon.AbstractDaemonService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/***
+/**
* Statistics of broker. It use two CountSet alternatively print statistics to
log.
*/
public class GroupCountService extends AbstractDaemonService implements
CountService {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/BrokerSamplePrint.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/BrokerSamplePrint.java
index 55fc036..15bc784 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/BrokerSamplePrint.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/BrokerSamplePrint.java
@@ -21,7 +21,7 @@ import java.io.IOException;
import org.apache.inlong.tubemq.corebase.utils.AbstractSamplePrint;
import org.slf4j.Logger;
-/***
+/**
* Compressed print broker's statistics.
*/
public class BrokerSamplePrint extends AbstractSamplePrint {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtils.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtils.java
index bd5e7ed..5c1eabd 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtils.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtils.java
@@ -29,7 +29,7 @@ import
org.apache.inlong.tubemq.corebase.utils.MessageFlagUtils;
import org.apache.inlong.tubemq.corebase.utils.TStringUtils;
import org.apache.inlong.tubemq.server.broker.stats.CountItem;
-/***
+/**
* Storage util. Used for data and index file storage format.
*/
public class DataStoreUtils {
@@ -103,15 +103,15 @@ public class DataStoreUtils {
return nf.format(offset) + fileSuffix;
}
- /***
+ /**
* Convert inner message to protobuf format, then reply to client.
*
- * @param dataBuffer
- * @param dataTotalSize
- * @param countMap
- * @param statisKeyBase
- * @param sBuilder
- * @return
+ * @param dataBuffer the raw stored data
+ * @param dataTotalSize the data size
+ * @param countMap the statistics map
+ * @param statisKeyBase the statistics key prefix
+ * @param sBuilder the string buffer
+ * @return the converted messages
*/
public static ClientBroker.TransferedMessage getTransferMsg(final
ByteBuffer dataBuffer, int dataTotalSize,
final
HashMap<String, CountItem> countMap,
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DiskSamplePrint.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DiskSamplePrint.java
index e30c30a..72a33f1 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DiskSamplePrint.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DiskSamplePrint.java
@@ -21,7 +21,7 @@ import java.io.IOException;
import org.apache.inlong.tubemq.corebase.utils.AbstractSamplePrint;
import org.slf4j.Logger;
-/***
+/**
* Compressed print disk exception's statistics.
*/
public class DiskSamplePrint extends AbstractSamplePrint {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/web/BrokerAdminServlet.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/web/BrokerAdminServlet.java
index 12b81f2..26747f6 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/web/BrokerAdminServlet.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/web/BrokerAdminServlet.java
@@ -50,7 +50,7 @@ import
org.apache.inlong.tubemq.server.common.TServerConstants;
import org.apache.inlong.tubemq.server.common.fielddef.WebFieldDef;
import org.apache.inlong.tubemq.server.common.utils.WebParameterUtils;
-/***
+/**
* Broker's web servlet. Used for admin operation, like query consumer's
status etc.
*/
public class BrokerAdminServlet extends AbstractWebHandler {
@@ -109,7 +109,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
"adminQueryGroupHistoryOffSet", false);
}
- /***
+ /**
* Query all API methods supported by this version.
*
* @param req request
@@ -130,7 +130,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuffer.append("],\"totalCnt\":").append(index).append("}");
}
- /***
+ /**
* Query broker's all consumer info.
*
* @param req request
@@ -218,7 +218,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
strBuff.append("],\"totalCnt\":").append(index).append("}");
}
- /***
+ /**
* Query broker's all message store info.
*
* @param req request
@@ -284,7 +284,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuilder.append("],\"totalCnt\":").append(recordId).append("}");
}
- /***
+ /**
* Get memory store status info.
*
* @param req request
@@ -342,7 +342,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuffer.append("],\"totalCount\":").append(recordId).append("}");
}
- /***
+ /**
* Manual set offset.
*
* @param req request
@@ -430,7 +430,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
}
}
- /***
+ /**
* Query snapshot message set.
*
* @param req request
@@ -474,7 +474,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
.getMessageSnapshot(topicName, partitionId, msgCount,
filterCondStrSet, sBuffer);
}
- /***
+ /**
* Query consumer group offset.
*
* @param req request
@@ -563,7 +563,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuffer.append("}");
}
- /***
+ /**
* Query the consumed partition information of online consumer.
*
* @param req request
@@ -590,7 +590,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuffer.append("],\"totalCnt\":").append(totalCnt).append("}");
}
- /***
+ /**
* Query topic's publish info on the Broker.
*
* @param req request
@@ -634,7 +634,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuffer.append("],\"dataCount\":").append(totalCnt).append("}");
}
- /***
+ /**
* Query all consumer groups booked on the Broker.
*
* @param req request
@@ -695,7 +695,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuffer.append("],\"dataCount\":").append(totalCnt).append("}");
}
- /***
+ /**
* Query consumer group offset.
*
* @param req request
@@ -769,7 +769,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuffer.append("],\"totalCnt\":").append(totalCnt).append("}");
}
- /***
+ /**
* Query consumer group history offset by timestamp.
*
* @param req request
@@ -842,7 +842,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
WebParameterUtils.buildSuccessWithDataRetEnd(sBuffer, totalCnt);
}
- /***
+ /**
* Query group's offset records stored in broker.
*
* @param msgStore history offset store
@@ -937,7 +937,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
}
}
- /***
+ /**
* Add or Modify consumer group offset.
*
* @param req request
@@ -999,7 +999,7 @@ public class BrokerAdminServlet extends AbstractWebHandler {
sBuffer.append("{\"result\":true,\"errCode\":0,\"errMsg\":\"OK\"}");
}
- /***
+ /**
* Clone consume group offset, clone A group's offset to other group.
*
* @param req request
@@ -1064,7 +1064,7 @@ public class BrokerAdminServlet extends
AbstractWebHandler {
sBuffer.append("{\"result\":true,\"errCode\":0,\"errMsg\":\"OK\"}");
}
- /***
+ /**
* Remove consume group offset.
*
* @param req request
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/web/WebServer.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/web/WebServer.java
index 4c70874..0c085bc 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/web/WebServer.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/web/WebServer.java
@@ -23,7 +23,7 @@ import org.apache.inlong.tubemq.server.broker.TubeBroker;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
-/***
+/**
* Broker's http server.
*/
public class WebServer implements Server {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/common/paramcheck/PBParameterUtils.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/common/paramcheck/PBParameterUtils.java
index a677598..0ec21bf 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/common/paramcheck/PBParameterUtils.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/common/paramcheck/PBParameterUtils.java
@@ -134,6 +134,15 @@ public class PBParameterUtils {
return retResult;
}
+ /**
+ * Check the validity of the bootstrap Offset information specified by the
consumer.
+ *
+ * @param csmType the topic list to be checked.
+ * @param reqTopicSet the subscribed topic set
+ * @param requiredParts the specified partitionKey-bootstrap offset map
+ * @param strBuffer the string buffer used to construct the result
+ * @return the check result
+ */
public static ParamCheckResult checkConsumerOffsetSetInfo(ConsumeType
csmType,
final
Set<String> reqTopicSet,
final String
requiredParts,
@@ -188,6 +197,17 @@ public class PBParameterUtils {
return retResult;
}
+ /**
+ * Check the validity of consumer parameters
+ * which specify partition boostrap Offset and use server-side balancing.
+ *
+ * @param inConsumerInfo the consumer information
+ * @param masterConfig the master configure
+ * @param defMetaDataManager the cluster meta information
+ * @param brokerRunManager the broker running information
+ * @param strBuffer the string buffer used to construct the
result
+ * @return the check result
+ */
public static ParamCheckResult checkConsumerInputInfo(ConsumerInfo
inConsumerInfo,
MasterConfig
masterConfig,
MetaDataManager
defMetaDataManager,
@@ -231,7 +251,7 @@ public class PBParameterUtils {
brokerRunManager.getSubTopicMaxBrokerCount(inConsumerInfo.getTopicSet());
int curBClientRate = (int) Math.floor(maxBrokerCount /
inConsumerInfo.getSourceCount());
if (curBClientRate > allowRate) {
- int minClientCnt = (int) (maxBrokerCount / allowRate);
+ int minClientCnt = maxBrokerCount / allowRate;
if (maxBrokerCount % allowRate != 0) {
minClientCnt += 1;
}
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/nodemanage/nodeconsumer/ConsumeGroupInfo.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/nodemanage/nodeconsumer/ConsumeGroupInfo.java
index f3d0ac3..b9823a6 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/nodemanage/nodeconsumer/ConsumeGroupInfo.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/nodemanage/nodeconsumer/ConsumeGroupInfo.java
@@ -217,6 +217,11 @@ public class ConsumeGroupInfo {
}
}
+ /**
+ * Get the client nodes that need to be balanced
+ *
+ * @return the client nodes
+ */
public RebProcessInfo getNeedBalanceNodes() {
List<String> needProcessList = new ArrayList<>();
List<String> needEscapeList = new ArrayList<>();
@@ -243,6 +248,11 @@ public class ConsumeGroupInfo {
return new RebProcessInfo(needProcessList, needEscapeList);
}
+ /**
+ * Update the client nodes status that has completed balancing
+ *
+ * @param processList the client nodes that has completed balancing
+ */
public void setBalanceNodeProcessed(List<String> processList) {
if (processList == null
|| processList.isEmpty()
@@ -269,22 +279,27 @@ public class ConsumeGroupInfo {
csmCtrlId.set(System.currentTimeMillis());
}
- public void updCsmTopicMetaInfo(Map<String, String> result) {
+ /**
+ * Update topic meta information
+ *
+ * @param newMetaInfoMap the newly acquired topic metadata
+ */
+ public void updCsmTopicMetaInfo(Map<String, String> newMetaInfoMap) {
lastMetaInfoFreshTime.set(System.currentTimeMillis());
- if (result == null || result.isEmpty()) {
+ if (newMetaInfoMap == null || newMetaInfoMap.isEmpty()) {
return;
}
String newConfig;
String curCOnfig;
boolean isChanged = false;
- Set<String> newTopics = result.keySet();
+ Set<String> newTopics = newMetaInfoMap.keySet();
Set<String> curTopics = topicMetaInfoMap.keySet();
if (newTopics.size() != curTopics.size()
|| !newTopics.containsAll(curTopics)) {
isChanged = true;
} else {
for (String topicKey : newTopics) {
- newConfig = result.get(topicKey);
+ newConfig = newMetaInfoMap.get(topicKey);
curCOnfig = topicMetaInfoMap.get(topicKey);
if (newConfig == null) {
continue;
@@ -297,7 +312,7 @@ public class ConsumeGroupInfo {
}
if (isChanged) {
for (String newTopic : newTopics) {
- topicMetaInfoMap.put(newTopic, result.get(newTopic));
+ topicMetaInfoMap.put(newTopic, newMetaInfoMap.get(newTopic));
}
topicMetaInfoId.set(System.currentTimeMillis());
}
@@ -632,7 +647,7 @@ public class ConsumeGroupInfo {
}
// check the topic conditions of consumption
boolean isCondEqual = true;
- if (topicConditions == null || topicConditions.isEmpty()) {
+ if (topicConditions.isEmpty()) {
if (!inConsumer.getTopicConditions().isEmpty()) {
isCondEqual = false;
sBuffer.append("[Inconsistency subscribe]
").append(inConsumer.getConsumerId())
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/nodemanage/nodeconsumer/ConsumerEventManager.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/nodemanage/nodeconsumer/ConsumerEventManager.java
index 1966480..0b5e80a 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/nodemanage/nodeconsumer/ConsumerEventManager.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/nodemanage/nodeconsumer/ConsumerEventManager.java
@@ -118,8 +118,8 @@ public class ConsumerEventManager {
* Removes and returns the first consumer event from event map
* disconnect event have priority over connect event
*
- * @param consumerId
- * @return the first consumer removed from the event map
+ * @param consumerId the consumer id that need removed
+ * @return the first consumer removed from the event map
*/
public ConsumerEvent removeFirst(String consumerId) {
ConsumerEvent event = null;
@@ -163,6 +163,11 @@ public class ConsumerEventManager {
return unfinishedCount.get();
}
+ /**
+ * Update the rounds of consumer groups dealing with balancing tasks
+ *
+ * @param groupHasUnfinishedEvent the consumer groups dealing with
balancing tasks
+ */
public void updateUnfinishedCountMap(Set<String> groupHasUnfinishedEvent) {
if (groupHasUnfinishedEvent.isEmpty()) {
groupUnfinishedCountMap.clear();
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/utils/Chore.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/utils/Chore.java
index 5fc378f..b465ddc 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/utils/Chore.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/utils/Chore.java
@@ -28,7 +28,7 @@ import org.slf4j.LoggerFactory;
* abstract class provides while loop and sleeping facility. If an unhandled
exception, the threads
* exit is logged. Implementers just need to add checking if there is work to
be done and if so, do
* it.
- * <p/>
+ *
* Don't subclass Chore if the task relies on being woken up for something to
do, such as an entry
* being added to a queue, etc.
*
@@ -40,6 +40,8 @@ public abstract class Chore extends HasThread {
private final Sleeper sleeper;
/**
+ * Initial Chore object
+ * @param name the chore thread name
* @param p Period at which we should run. Will be adjusted
appropriately should we find
* work and it takes time to complete.
* @param stopper When {@link Stoppable#isStopped()} is true, this thread
will cleanup and exit
@@ -52,7 +54,7 @@ public abstract class Chore extends HasThread {
}
/**
- * @see Thread#run()
+ * The chore thread processing logic
*/
@Override
public void run() {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/common/BaseResult.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/common/BaseResult.java
index c5dabea..024098b 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/common/BaseResult.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/common/BaseResult.java
@@ -30,8 +30,8 @@ import org.apache.inlong.tubemq.corebase.utils.TStringUtils;
/**
* Paging algorithm package.
- * * <p/>
- * * Pagination must be set: TotalItem (the total number of bars), the
default is 0,
+ *
+ * Pagination must be set: TotalItem (the total number of bars), the default
is 0,
* should be set in dao PageSize (number of pages per page), should be set in
the web
* layer QueryBase defaults to 20, subclasses can be overwritten
getDefaultPageSize()
* Modify CurrentPage (current page), default is 1, home page, should be set
in the
@@ -39,15 +39,15 @@ import org.apache.inlong.tubemq.corebase.utils.TStringUtils;
* (the current page starts recording position, counting from 1) PageLastItem
* (current page last recording position) On the page, the number of pages
displayed
* per page should be: lines , the current page name should be: page
- * * <p/>
- * * Add the render link function at the same time,
+ *
+ * Add the render link function at the same time,
* the subclass overrides the getParameters method and returns valid
parameters.
*/
public class BaseResult implements Serializable {
private static final long serialVersionUID = 8807356835558347735L;
- private static final Integer defaultPageSize = new Integer(20);
- private static final Integer defaultFirstPage = new Integer(1);
- private static final Integer defaultTotalItem = new Integer(0);
+ private static final Integer defaultPageSize = 20;
+ private static final Integer defaultFirstPage = 1;
+ private static final Integer defaultTotalItem = 0;
/**
* max page size
*/
@@ -70,8 +70,12 @@ public class BaseResult implements Serializable {
/**
* parse date
+ *
+ * @param dateTime the string date time
+ * @param format the date format
+ * @param def the defalut date value
*/
- public static final Date parseDate(String dateTime, String format, Date
def) {
+ public static Date parseDate(String dateTime, String format, Date def) {
Date date = def;
try {
DateFormat formatter = new SimpleDateFormat(format);
@@ -92,7 +96,7 @@ public class BaseResult implements Serializable {
*
* @return Encoded string
*/
- public static final String jsEncode(String str) {
+ public static String jsEncode(String str) {
if (null == str) {
return null;
}
@@ -125,7 +129,7 @@ public class BaseResult implements Serializable {
*
* @return Encoded string
*/
- public static final String jsDecode(String str) {
+ public static String jsDecode(String str) {
if (null == str) {
return null;
}
@@ -171,6 +175,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get default page size.
+ *
* @return Returns the defaultPageSize.
*/
protected Integer getDefaultPageSize() {
@@ -206,6 +212,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get current page value
+ *
* @return Returns the currentPage.
*/
public Integer getCurrentPage() {
@@ -217,10 +225,12 @@ public class BaseResult implements Serializable {
}
/**
+ * Set current page value
+ *
* @param cPage The currentPage to set.
*/
public void setCurrentPage(Integer cPage) {
- if ((cPage == null) || (cPage.intValue() <= 0)) {
+ if ((cPage == null) || (cPage <= 0)) {
this.currentPage = null;
} else {
this.currentPage = cPage;
@@ -247,6 +257,7 @@ public class BaseResult implements Serializable {
}
/**
+ * Get page size value
* @return Returns the pageSize.
*/
public Integer getPageSize() {
@@ -258,11 +269,13 @@ public class BaseResult implements Serializable {
}
/**
+ * Set page size value
+ *
* @param pSize The pageSize to set.
*/
public void setPageSize(Integer pSize) {
- if ((pSize == null) || (pSize.intValue() < 0)) {
+ if ((pSize == null) || (pSize < 0)) {
this.pageSize = null;
} else if (pSize > MAX_PAGE_SIZE || pSize < 1) {
throw new IllegalArgumentException("The number of displayed pages
per page ranges from 1~" + MAX_PAGE_SIZE);
@@ -294,8 +307,10 @@ public class BaseResult implements Serializable {
}
/**
- * @param pageSizeString
- * @return
+ * Determine if the string is blank
+ *
+ * @param pageSizeString pagesize value
+ * @return result
*/
private boolean isBlankString(String pageSizeString) {
if (pageSizeString == null) {
@@ -307,6 +322,7 @@ public class BaseResult implements Serializable {
}
/**
+ * Get total item value
* @return Returns the totalItem.
*/
public Integer getTotalItem() {
@@ -319,6 +335,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Set total item value
+ *
* @param tItem The totalItem to set.
*/
public void setTotalItem(Integer tItem) {
@@ -368,14 +386,12 @@ public class BaseResult implements Serializable {
int assumeLast = pgSize * cPage;
int totalItem = getTotalItem().intValue();
- if (assumeLast > totalItem) {
- return totalItem;
- } else {
- return assumeLast;
- }
+ return Math.min(assumeLast, totalItem);
}
/**
+ * Get end row value
+ *
* @return Returns the endRow.
*/
public int getEndRow() {
@@ -383,6 +399,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Set end row value
+ *
* @param endRow The endRow to set.
*/
public void setEndRow(int endRow) {
@@ -390,6 +408,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get start row value
+ *
* @return Returns the startRow.
*/
public int getStartRow() {
@@ -397,6 +417,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Set start row value
+ *
* @param startRow The startRow to set.
*/
public void setStartRow(int startRow) {
@@ -421,6 +443,8 @@ public class BaseResult implements Serializable {
/**
* When the time is queried, the end time is 23:59:59
+ *
+ * @param dateString the sting date value
*/
protected String addDateEndPostfix(String dateString) {
if (TStringUtils.isBlank(dateString)) {
@@ -432,6 +456,8 @@ public class BaseResult implements Serializable {
/**
* When the time is queried, the start time is 00:00:00
+ *
+ * @param dateString the string date value
*/
protected String addDateStartPostfix(String dateString) {
if (TStringUtils.isBlank(dateString)) {
@@ -487,6 +513,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get ajax prefix value
+ *
* @return Returns the ajaxPrefix.
*/
public String getAjaxPrefix() {
@@ -494,6 +522,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Set ajax prefix
+ *
* @param ajaxPrefix The ajaxPrefix to set.
*/
public BaseResult setAjaxPrefix(String ajaxPrefix) {
@@ -502,6 +532,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get ajaxSuffix value
+ *
* @return Returns the ajaxSuffix.
*/
public String getAjaxSuffix() {
@@ -509,6 +541,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Set ajax suffix
+ *
* @param ajaxSuffix The ajaxSuffix to set.
*/
public BaseResult setAjaxSuffix(String ajaxSuffix) {
@@ -517,6 +551,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get charset
+ *
* @return Returns the charset.
*/
public String getCharset() {
@@ -524,6 +560,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Set charset
+ *
* @param charset The charset to set.
*/
public BaseResult setCharset(String charset) {
@@ -533,6 +571,8 @@ public class BaseResult implements Serializable {
/**
* Remove a parameter
+ *
+ * @param key the Key that need to be removed
*/
public BaseResult remove(Object key) {
if (null == this.removeObject) {
@@ -544,6 +584,9 @@ public class BaseResult implements Serializable {
/**
* Temporarily modify the value of a parameter
+ *
+ * @param key the Key that need to be modified
+ * @param val the new value
*/
public BaseResult replace(Object key, Object val) {
if (null != key && null != val) {
@@ -556,6 +599,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get source
+ *
* @return Returns the from.
*/
public String getFrom() {
@@ -563,6 +608,8 @@ public class BaseResult implements Serializable {
}
/**
+ * set source
+ *
* @param from The from to set.
*/
public void setFrom(String from) {
@@ -570,6 +617,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get whether to escape
+ *
* @return Returns the escape.
*/
public boolean isEscape() {
@@ -577,6 +626,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Set escape
+ *
* @param escape The escape to set.
*/
public BaseResult setEscape(boolean escape) {
@@ -585,6 +636,8 @@ public class BaseResult implements Serializable {
}
/**
+ * Get whether to escape js
+ *
* @return Returns the jsEscape.
*/
public final boolean isJsEscape() {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/simplemvc/RequestDispatcher.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/simplemvc/RequestDispatcher.java
index 57f297e..fa387af 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/simplemvc/RequestDispatcher.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/simplemvc/RequestDispatcher.java
@@ -51,7 +51,7 @@ public class RequestDispatcher {
/**
* Construct RequestDispatcher from web config
*
- * @param config
+ * @param config the web configure object
*/
public RequestDispatcher(WebConfig config) {
this.config = config;
@@ -141,9 +141,9 @@ public class RequestDispatcher {
/**
* Build action key as "type/target"
*
- * @param type
- * @param target
- * @return
+ * @param type the type value
+ * @param target the target value
+ * @return the key
*/
public String getActionKey(String type, String target) {
return new StringBuilder(256).append(type)
@@ -153,9 +153,9 @@ public class RequestDispatcher {
/**
* Build template name as "type/target.vm"
*
- * @param type
- * @param target
- * @return
+ * @param type the type value
+ * @param target the target value
+ * @return the result
*/
public String getTemplateName(String type, String target) {
return new StringBuilder(256).append(type)
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/simplemvc/conf/ConfigFileParser.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/simplemvc/conf/ConfigFileParser.java
index acc93ba..2c7ea55 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/simplemvc/conf/ConfigFileParser.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/web/simplemvc/conf/ConfigFileParser.java
@@ -60,6 +60,11 @@ public class ConfigFileParser {
return parse();
}
+ /**
+ * Parse website configure file information
+ *
+ * @return the website configure object
+ */
public WebConfig parse() throws Exception {
SAXReader reader = new SAXReader();
WebConfig config = new WebConfig();
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/BdbGroupAdmin.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/BdbGroupAdmin.java
index 39239ae..d9f9616 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/BdbGroupAdmin.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/BdbGroupAdmin.java
@@ -45,9 +45,15 @@ import java.util.Set;
* until it transitions to the Replica state. If you attempt to remove an
active Master, a MasterStateException
* is thrown.
*/
-
public class BdbGroupAdmin {
+ /**
+ * Remove a node from a BDB-JE replication group.
+ * @param args Startup parameter array, including the following parts:
+ * The 1st parameter is replication group name;
+ * The 2nd parameter is helperHost address, format is
ip:port;
+ * The 3nd parameter is need remove node name.
+ */
public static void main(final String[] args) throws Exception {
if (args == null || args.length != 3) {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/CliUtils.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/CliUtils.java
index 9398c8a..74d12e9 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/CliUtils.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/CliUtils.java
@@ -26,8 +26,20 @@ import org.apache.commons.cli.Options;
import org.apache.inlong.tubemq.corebase.rv.ProcessResult;
import org.apache.inlong.tubemq.corebase.utils.TStringUtils;
+/**
+ * Cli related utils
+ *
+ */
public class CliUtils {
+ /**
+ * Get and verify the configuration file path.
+ * @param args Call parameter array,
+ * the relevant parameters are dynamic mode, which is
parsed by CommandLine.
+ * @param result the find result
+ *
+ * @return whether success or failure
+ */
public static boolean getConfigFilePath(final String[] args, ProcessResult
result) {
// build file option
Options options = new Options();
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/StoreRepairAdmin.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/StoreRepairAdmin.java
index 82f78ba..41a19f3 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/StoreRepairAdmin.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/StoreRepairAdmin.java
@@ -43,10 +43,21 @@ import
org.apache.inlong.tubemq.server.common.utils.FileUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Topic Storage File(s) Repair Tool
+ *
+ */
public class StoreRepairAdmin {
private static final Logger logger =
LoggerFactory.getLogger(StoreRepairAdmin.class);
+ /**
+ * Repair topic storage files, rebuild index files by existing data files
+ * @param args Startup parameter array, including the following parts:
+ * The 1st parameter is the path of topic storage files;
+ * The 2nd parameter is the topic name(s) that needs to be
repaired,
+ * if not specified, all topics in the storage path will
be repaired.
+ */
public static void main(final String[] args) throws Exception {
if (args == null || args.length < 1) {
System.out.println(
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliAbstractBase.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliAbstractBase.java
index d4519f4..aab466a 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliAbstractBase.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliAbstractBase.java
@@ -30,7 +30,7 @@ public abstract class CliAbstractBase {
protected final String commandName;
protected Options options = new Options();
protected CommandLineParser parser = new DefaultParser();
- private HelpFormatter formatter = new HelpFormatter();
+ private final HelpFormatter formatter = new HelpFormatter();
public CliAbstractBase(String commandName) {
this.commandName = commandName;
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliBrokerAdmin.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliBrokerAdmin.java
index 00f3b07..c19f27b 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliBrokerAdmin.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliBrokerAdmin.java
@@ -29,7 +29,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This class is use to process CLI Broker Admin process.
+ * This class is use to process CLI Broker Admin process for script
#{bin/tubemq-broker-admin.sh}.
*
*
*/
@@ -56,6 +56,11 @@ public class CliBrokerAdmin extends CliAbstractBase {
}
+ /**
+ * Call the broker's HTTP API by the tubemq-broker-admin.sh script
+ * @param args Call parameter array,
+ * the relevant parameters are dynamic mode, which is
parsed by CommandLine.
+ */
public boolean processParams(String[] args) throws Exception {
// parse parameters and check value
CommandLine cli = parser.parse(options, args);
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliConsumer.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliConsumer.java
index 8417863..e88dc7f 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliConsumer.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliConsumer.java
@@ -50,9 +50,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This class is use to process CLI Consumer process.
- *
- *
+ * This class is use to process CLI Consumer process for script
#{bin/tubemq-consumer-test.sh}.
*/
public class CliConsumer extends CliAbstractBase {
@@ -184,7 +182,9 @@ public class CliConsumer extends CliAbstractBase {
return true;
}
- // initial tubemq client order by caller required
+ /**
+ * Initializes the TubeMQ consumer client(s) with the specified
requirements.
+ */
public void initTask() throws Exception {
// initial consumer configure
ConsumerConfig consumerConfig =
@@ -369,6 +369,11 @@ public class CliConsumer extends CliAbstractBase {
}
}
+ /**
+ * Consume messages called by the tubemq-consumer-test.sh script.
+ * @param args Call parameter array,
+ * the relevant parameters are dynamic mode, which is
parsed by CommandLine.
+ */
public static void main(String[] args) {
CliConsumer cliConsumer = new CliConsumer();
try {
diff --git
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliProducer.java
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliProducer.java
index 9661ade..d98d7a9 100644
---
a/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliProducer.java
+++
b/inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliProducer.java
@@ -46,7 +46,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This class is use to process CLI Producer process.
+ * This class is use to process CLI Producer process for script
#{bin/tubemq-producer-test.sh}.
*
*
*/
@@ -179,7 +179,9 @@ public class CliProducer extends CliAbstractBase {
return true;
}
- // initial tubemq client order by caller required
+ /**
+ * Initializes the TubeMQ producer client(s) with the specified
requirements.
+ */
public void initTask() throws Exception {
// initial client configure
TubeClientConfig clientConfig = new TubeClientConfig(masterServers);
@@ -312,6 +314,11 @@ public class CliProducer extends CliAbstractBase {
}
}
+ /**
+ * Produce messages called by the tubemq-producer-test.sh script.
+ * @param args Call parameter array,
+ * the relevant parameters are dynamic mode, which is
parsed by CommandLine.
+ */
public static void main(String[] args) {
CliProducer cliProducer = new CliProducer();
try {
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 7e9dd7e..ea2813e 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
@@ -23,7 +23,7 @@ import java.util.Map;
import org.junit.Assert;
import org.junit.Test;
-/***
+/**
* BrokerMetadataManage test
*/
public class BrokerMetadataManagerTest {
diff --git
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentListTest.java
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentListTest.java
index 0c38660..b2c6fd8 100644
---
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentListTest.java
+++
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentListTest.java
@@ -24,7 +24,7 @@ import
org.apache.inlong.tubemq.server.broker.utils.DataStoreUtils;
import org.junit.Assert;
import org.junit.Test;
-/***
+/**
* FileSegmentList test
*/
public class FileSegmentListTest {
diff --git
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentTest.java
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentTest.java
index 959fd83..29c7251 100644
---
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentTest.java
+++
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/FileSegmentTest.java
@@ -21,7 +21,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
-/***
+/**
* FileSegment test.
*/
public class FileSegmentTest {
diff --git
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStoreTest.java
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStoreTest.java
index c08ab34..651968d 100644
---
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStoreTest.java
+++
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/msgstore/mem/MsgMemStoreTest.java
@@ -22,7 +22,7 @@ import java.nio.ByteBuffer;
import org.apache.inlong.tubemq.server.common.utils.AppendResult;
import org.junit.Test;
-/***
+/**
* MsgMemStore test.
*/
public class MsgMemStoreTest {
diff --git
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/stats/GroupCountServiceTest.java
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/stats/GroupCountServiceTest.java
index d2ef145..b9e80fe 100644
---
a/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/stats/GroupCountServiceTest.java
+++
b/inlong-tubemq/tubemq-server/src/test/java/org/apache/inlong/tubemq/server/broker/stats/GroupCountServiceTest.java
@@ -21,7 +21,7 @@ import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
-/***
+/**
* GroupCountService test.
*/
public class GroupCountServiceTest {
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 c240064..6cd5b37 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
@@ -21,7 +21,7 @@ import java.nio.ByteBuffer;
import org.junit.Assert;
import org.junit.Test;
-/***
+/**
* DataStoreUtils test.
*/
public class DataStoreUtilsTest {
