This is an automated email from the ASF dual-hosted git repository.
dockerzhang 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 38c525b [INLONG-2045] sort-sdk support Prometheus metrics (#2114)
38c525b is described below
commit 38c525b4990882fc209411fba934e8afa98effaa
Author: wardli <[email protected]>
AuthorDate: Fri Jan 7 19:23:41 2022 +0800
[INLONG-2045] sort-sdk support Prometheus metrics (#2114)
---
inlong-sdk/sort-sdk/pom.xml | 6 ++
.../apache/inlong/sdk/sort/api/MetricReporter.java | 2 +-
.../inlong/sdk/sort/api/SortClientConfig.java | 9 ++
.../inlong/sdk/sort/impl/MetricReporterImpl.java | 70 +++++++++++++-
.../inlong/sdk/sort/impl/SortClientImpl.java | 2 +-
.../inlong/sdk/sort/metrics/SortSdkMetricItem.java | 104 +++++++++++++++++++++
.../metrics/SortSdkPrometheusMetricListener.java | 95 +++++++++++++++++++
.../sdk/sort/stat/SortClientStateCounter.java | 82 ++++++++--------
.../apache/inlong/sdk/sort/stat/StatManager.java | 35 +++----
9 files changed, 342 insertions(+), 63 deletions(-)
diff --git a/inlong-sdk/sort-sdk/pom.xml b/inlong-sdk/sort-sdk/pom.xml
index 51e388b..c061b10 100644
--- a/inlong-sdk/sort-sdk/pom.xml
+++ b/inlong-sdk/sort-sdk/pom.xml
@@ -112,6 +112,12 @@
<version>2.23.0</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>inlong-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
</dependencies>
</project>
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/MetricReporter.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/MetricReporter.java
index b6aaf4e..6479c2b 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/MetricReporter.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/MetricReporter.java
@@ -19,7 +19,7 @@ package org.apache.inlong.sdk.sort.api;
public interface MetricReporter {
- boolean report(String monitorName, String[] keys, double[] values);
+ boolean report(String monitorName, String[] keys, long[] values);
void close();
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
index 1b8d009..b50eace 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
@@ -44,6 +44,7 @@ public class SortClientConfig implements Serializable {
private int updateMetaDataIntervalSec = 10;
private int ackTimeoutSec = 10;
private volatile boolean stopConsume = false;
+ private boolean isPrometheusEnabled = true;
public SortClientConfig(String sortTaskId, String sortClusterName,
InLongTopicChangeListener assignmentsListener,
ConsumeStrategy consumeStrategy, String localIp) {
@@ -228,6 +229,14 @@ public class SortClientConfig implements Serializable {
this.managerApiVersion = managerApiVersion;
}
+ public boolean isPrometheusEnabled() {
+ return isPrometheusEnabled;
+ }
+
+ public void setPrometheusEnabled(boolean prometheusEnabled) {
+ isPrometheusEnabled = prometheusEnabled;
+ }
+
/**
* ConsumeStrategy
*/
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/MetricReporterImpl.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/MetricReporterImpl.java
index fba1223..2dcfe49 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/MetricReporterImpl.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/MetricReporterImpl.java
@@ -17,12 +17,30 @@
package org.apache.inlong.sdk.sort.impl;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.inlong.sdk.sort.api.MetricReporter;
+import org.apache.inlong.sdk.sort.api.SortClientConfig;
+import org.apache.inlong.sdk.sort.metrics.SortSdkMetricItem;
+import org.apache.inlong.sdk.sort.metrics.SortSdkPrometheusMetricListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class MetricReporterImpl implements MetricReporter {
+ private static final Logger LOG =
LoggerFactory.getLogger(MetricReporterImpl.class);
+ private SortSdkPrometheusMetricListener sortSdkPrometheusMetricListener;
+ private SortClientConfig sortClientConfig;
+
+ public MetricReporterImpl(SortClientConfig sortClientConfig) {
+ this.sortClientConfig = sortClientConfig;
+ if (sortClientConfig.isPrometheusEnabled()) {
+ initPrometheusMetricListener();
+ }
+ }
+
/**
- * report metric by JMX
+ * report metric
*
* @param monitorName String
* @param keys metric key
@@ -30,8 +48,20 @@ public class MetricReporterImpl implements MetricReporter {
* @return true/false
*/
@Override
- public boolean report(String monitorName, String[] keys, double[] values) {
- return false;
+ public boolean report(String monitorName, String[] keys, long[] values) {
+
+ if (sortClientConfig.isPrometheusEnabled()) {
+ if (sortSdkPrometheusMetricListener != null) {
+ Map<String, Long> prometheusMetricVals =
getPrometheusMetricVals(values);
+ if (prometheusMetricVals != null) {
+ sortSdkPrometheusMetricListener.snapshot(monitorName,
prometheusMetricVals);
+ } else {
+ LOG.warn("sortSdkPrometheusMetricListener is null! now
init it");
+ initPrometheusMetricListener();
+ }
+ }
+ }
+ return true;
}
/**
@@ -41,4 +71,38 @@ public class MetricReporterImpl implements MetricReporter {
public void close() {
}
+
+ private void initPrometheusMetricListener() {
+ sortSdkPrometheusMetricListener = new
SortSdkPrometheusMetricListener(sortClientConfig.getSortTaskId());
+ }
+
+ private Map<String, Long> getPrometheusMetricVals(long[] values) {
+ if (values.length > 15) {
+ Map<String, Long> metricValueMap = new HashMap<>();
+ // consume
+ metricValueMap.put(SortSdkMetricItem.M_CONSUME_SIZE, values[0]);
+ metricValueMap.put(SortSdkMetricItem.M_CONSUME_MSG_COUNT,
values[1]);
+ // callback
+ metricValueMap.put(SortSdkMetricItem.M_CALL_BACK_COUNT, values[2]);
+ metricValueMap.put(SortSdkMetricItem.M_CALL_BACK_DONE_COUNT,
values[3]);
+ metricValueMap.put(SortSdkMetricItem.M_CALL_BACK_TIME_COST,
values[4]);
+ metricValueMap.put(SortSdkMetricItem.M_CALL_BACK_FAIL_COUNT,
values[5]);
+ // topic
+ metricValueMap.put(SortSdkMetricItem.M_TOPIC_ONLINE_COUNT,
values[6]);
+ metricValueMap.put(SortSdkMetricItem.M_TOPIC_OFFLINE_COUNT,
values[7]);
+ // ack
+ metricValueMap.put(SortSdkMetricItem.M_ACK_FAIL_COUNT, values[8]);
+ metricValueMap.put(SortSdkMetricItem.M_ACK_SUCC_COUNT, values[9]);
+ // request manager
+ metricValueMap.put(SortSdkMetricItem.M_REQUEST_MANAGER_COUNT,
values[10]);
+ metricValueMap.put(SortSdkMetricItem.M_REQUEST_MANAGER_TIME_COST,
values[11]);
+ metricValueMap.put(SortSdkMetricItem.M_REQUEST_MANAGER_FAIL_COUNT,
values[12]);
+
metricValueMap.put(SortSdkMetricItem.M_REQUEST_MANAGER_CONF_CHANAGED_COUNT,
values[13]);
+
metricValueMap.put(SortSdkMetricItem.M_RQUEST_MANAGER_COMMON_ERROR_COUNT,
values[14]);
+
metricValueMap.put(SortSdkMetricItem.M_RQUEST_MANAGER_PARAM_ERROR_COUNT,
values[15]);
+ return metricValueMap;
+ }
+ return null;
+ }
+
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java
index 7f492b1..335d610 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java
@@ -49,7 +49,7 @@ public class SortClientImpl extends SortClient {
public SortClientImpl(SortClientConfig sortClientConfig) {
try {
this.sortClientConfig = sortClientConfig;
- this.context = new ClientContextImpl(this.sortClientConfig, new
MetricReporterImpl());
+ this.context = new ClientContextImpl(this.sortClientConfig, new
MetricReporterImpl(sortClientConfig));
this.inLongTopicManager = new InLongTopicManagerImpl(context,
new QueryConsumeConfigImpl(context));
} catch (Exception e) {
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/metrics/SortSdkMetricItem.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/metrics/SortSdkMetricItem.java
new file mode 100644
index 0000000..4dcbe95
--- /dev/null
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/metrics/SortSdkMetricItem.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sdk.sort.metrics;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.inlong.commons.config.metrics.CountMetric;
+import org.apache.inlong.commons.config.metrics.Dimension;
+import org.apache.inlong.commons.config.metrics.MetricDomain;
+import org.apache.inlong.commons.config.metrics.MetricItem;
+
+@MetricDomain(name = "SortSdk")
+public class SortSdkMetricItem extends MetricItem {
+
+ //Dimension
+ public static final String KEY_SORT_TASK_ID = "sortTaskId";
+ public static final String KEY_CLUSTER_ID = "clusterId";
+ public static final String KEY_TOPIC_ID = "topicId";
+ public static final String KEY_PARTITION_ID = "partitionId";
+
+ //CountMetric
+ //consume
+ public static final String M_CONSUME_SIZE = "consumeSize";
+ public static final String M_CONSUME_MSG_COUNT = "consumeMsgCount";
+ //callback
+ public static final String M_CALL_BACK_COUNT = "callbackCount";
+ public static final String M_CALL_BACK_DONE_COUNT = "callbackDoneCount";
+ public static final String M_CALL_BACK_TIME_COST = "callbakTimeCost";
+ public static final String M_CALL_BACK_FAIL_COUNT = "callbackFailCount";
+ //topic chanage
+ public static final String M_TOPIC_ONLINE_COUNT = "topicOnlineCount";
+ public static final String M_TOPIC_OFFLINE_COUNT = "topicOfflineCount";
+ //ack
+ public static final String M_ACK_FAIL_COUNT = "ackFailCount";
+ public static final String M_ACK_SUCC_COUNT = "ackSUCCCount";
+ //request manager
+ public static final String M_REQUEST_MANAGER_COUNT = "requestManagerCount";
+ public static final String M_REQUEST_MANAGER_TIME_COST =
"requestManagerTimeCost";
+ public static final String M_REQUEST_MANAGER_FAIL_COUNT =
"requestManagerFailCount";
+ public static final String M_REQUEST_MANAGER_CONF_CHANAGED_COUNT =
"requestManagerConfChanagedCount";
+ public static final String M_RQUEST_MANAGER_COMMON_ERROR_COUNT =
"requestManagerCommonErrorCount";
+ public static final String M_RQUEST_MANAGER_PARAM_ERROR_COUNT =
"requestManagerParamErrorCount";
+
+ @Dimension
+ public String sortTaskId;
+ @Dimension
+ public String clusterId;
+ @Dimension
+ public String topic;
+ @Dimension
+ public String partitionId;
+
+ @CountMetric
+ public AtomicLong consumeSize = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong consumeMsgCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong callbackCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong callbackDoneCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong callbackTimeCost = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong callbackFailCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong topicOnlineCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong topicOfflineCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong ackFailCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong ackSuccCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong requestManagerCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong requestManagerTimeCost = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong requestManagerFailCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong requestManagerConfChangedCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong requestManagerCommonErrorCount = new AtomicLong(0);
+ @CountMetric
+ public AtomicLong requestManagerParamErrorCount = new AtomicLong(0);
+
+ public SortSdkMetricItem(String sortTaskId) {
+ this.sortTaskId = sortTaskId;
+ }
+
+}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/metrics/SortSdkPrometheusMetricListener.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/metrics/SortSdkPrometheusMetricListener.java
new file mode 100644
index 0000000..2cd49bd
--- /dev/null
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/metrics/SortSdkPrometheusMetricListener.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sdk.sort.metrics;
+
+import static
org.apache.inlong.commons.config.metrics.MetricItemMBean.DOMAIN_SEPARATOR;
+import static
org.apache.inlong.commons.config.metrics.MetricRegister.JMX_DOMAIN;
+
+import java.lang.management.ManagementFactory;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SortSdkPrometheusMetricListener {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SortSdkPrometheusMetricListener.class);
+
+ private SortSdkMetricItem metricItem;
+ private Map<String, AtomicLong> metricValueMap = new ConcurrentHashMap<>();
+
+ public SortSdkPrometheusMetricListener(String sortTaskId) {
+ this.metricItem = new SortSdkMetricItem(sortTaskId);
+
+ final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ StringBuilder beanName = new StringBuilder();
+
beanName.append(JMX_DOMAIN).append(DOMAIN_SEPARATOR).append("type=SortSdkCounter");
+ String strBeanName = beanName.toString();
+ try {
+ ObjectName objName = new ObjectName(strBeanName);
+ mbs.registerMBean(metricItem, objName);
+ } catch (Exception ex) {
+ LOG.error("exception while register mbean:{},error:{}",
strBeanName, ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ }
+
+ // consume
+ metricValueMap.put(metricItem.M_CONSUME_SIZE, metricItem.consumeSize);
+ metricValueMap.put(metricItem.M_CONSUME_MSG_COUNT,
metricItem.consumeMsgCount);
+ // callback
+ metricValueMap.put(metricItem.M_CALL_BACK_COUNT,
metricItem.callbackCount);
+ metricValueMap.put(metricItem.M_CALL_BACK_DONE_COUNT,
metricItem.callbackDoneCount);
+ metricValueMap.put(metricItem.M_CALL_BACK_TIME_COST,
metricItem.callbackTimeCost);
+ metricValueMap.put(metricItem.M_CALL_BACK_FAIL_COUNT,
metricItem.callbackFailCount);
+ // topic
+ metricValueMap.put(metricItem.M_TOPIC_ONLINE_COUNT,
metricItem.topicOnlineCount);
+ metricValueMap.put(metricItem.M_TOPIC_OFFLINE_COUNT,
metricItem.topicOfflineCount);
+ // ack
+ metricValueMap.put(metricItem.M_ACK_FAIL_COUNT,
metricItem.ackFailCount);
+ metricValueMap.put(metricItem.M_ACK_SUCC_COUNT,
metricItem.ackSuccCount);
+ // request manager
+ metricValueMap.put(metricItem.M_REQUEST_MANAGER_COUNT,
metricItem.requestManagerCount);
+ metricValueMap.put(metricItem.M_REQUEST_MANAGER_TIME_COST,
metricItem.requestManagerTimeCost);
+ metricValueMap.put(metricItem.M_REQUEST_MANAGER_FAIL_COUNT,
metricItem.requestManagerFailCount);
+ metricValueMap.put(metricItem.M_REQUEST_MANAGER_CONF_CHANAGED_COUNT,
metricItem.requestManagerConfChangedCount);
+ metricValueMap.put(metricItem.M_RQUEST_MANAGER_COMMON_ERROR_COUNT,
metricItem.requestManagerCommonErrorCount);
+ metricValueMap.put(metricItem.M_RQUEST_MANAGER_PARAM_ERROR_COUNT,
metricItem.requestManagerParamErrorCount);
+
+ }
+
+ /**
+ * snapshot
+ *
+ * @param domain
+ * @param metrics
+ */
+ public void snapshot(String domain, Map<String, Long> metrics) {
+ for (Map.Entry<String, Long> entry : metrics.entrySet()) {
+ String fieldName = entry.getKey();
+ AtomicLong metricValue = this.metricValueMap.get(fieldName);
+ if (metricValue != null) {
+ long fieldValue = entry.getValue();
+ metricValue.addAndGet(fieldValue);
+ }
+ }
+ }
+
+}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
index f86381e..5fe1c75 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
@@ -22,22 +22,22 @@ import java.util.concurrent.atomic.AtomicLongArray;
public class SortClientStateCounter {
private final AtomicLongArray count = new AtomicLongArray(20);
- public String sortId;
- public String clusterId;
+ public String sortTaskId;
+ public String cacheClusterId;
public String topic;
public int partitionId;
/**
* SortClientStateCounter Constructor
*
- * @param sortId String
- * @param clusterId String
+ * @param sortTaskId String
+ * @param cacheClusterId String
* @param topic String
* @param partitionId int
*/
- public SortClientStateCounter(String sortId, String clusterId, String
topic, int partitionId) {
- this.sortId = sortId;
- this.clusterId = clusterId;
+ public SortClientStateCounter(String sortTaskId, String cacheClusterId,
String topic, int partitionId) {
+ this.sortTaskId = sortTaskId;
+ this.cacheClusterId = cacheClusterId;
this.topic = topic;
this.partitionId = partitionId;
}
@@ -48,7 +48,7 @@ public class SortClientStateCounter {
* @return SortClientStateCounter
*/
public SortClientStateCounter reset() {
- SortClientStateCounter counter = new SortClientStateCounter(sortId,
clusterId, topic, partitionId);
+ SortClientStateCounter counter = new
SortClientStateCounter(sortTaskId, cacheClusterId, topic, partitionId);
for (int i = 0, len = counter.count.length(); i < len; i++) {
counter.count.set(i, this.count.getAndSet(i, 0));
}
@@ -60,8 +60,8 @@ public class SortClientStateCounter {
*
* @return double[]
*/
- public double[] getStatvalue() {
- double[] vals = new double[this.count.length()];
+ public long[] getStatvalue() {
+ long[] vals = new long[this.count.length()];
for (int i = 0, len = this.count.length(); i < len; i++) {
vals[i] = this.count.get(i);
}
@@ -80,133 +80,133 @@ public class SortClientStateCounter {
}
/**
- * count callbak times
+ * count receive event num
*
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addCallbackTimes(int num) {
+ public SortClientStateCounter addMsgCount(long num) {
count.getAndAdd(1, num);
return this;
}
/**
- * count callbak time cost
+ * count callbak times
*
- * @param num long
+ * @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addCallbackTimeCost(long num) {
+ public SortClientStateCounter addCallbackTimes(long num) {
count.getAndAdd(2, num);
return this;
}
/**
- * count topic online times
+ * count callbak done times
*
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addTopicOnlineTimes(int num) {
+ public SortClientStateCounter addCallbackDoneTimes(long num) {
count.getAndAdd(3, num);
return this;
}
/**
- * count topic offline times
+ * count callbak time cost
*
- * @param num int
+ * @param num long
* @return SortClientStateCounter
*/
- public SortClientStateCounter addTopicOfflineTimes(int num) {
+ public SortClientStateCounter addCallbackTimeCost(long num) {
count.getAndAdd(4, num);
return this;
}
/**
- * count request manager times
+ * count callbak error times
*
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addRequestManagerTimes(int num) {
+ public SortClientStateCounter addCallbackErrorTimes(long num) {
count.getAndAdd(5, num);
return this;
}
/**
- * count request manager time cost
+ * count topic online times
*
- * @param num long
+ * @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addRequestManagerTimeCost(long num) {
+ public SortClientStateCounter addTopicOnlineTimes(long num) {
count.getAndAdd(6, num);
return this;
}
/**
- * count request manager fail times
+ * count topic offline times
*
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addRequestManagerFailTimes(int num) {
+ public SortClientStateCounter addTopicOfflineTimes(long num) {
count.getAndAdd(7, num);
return this;
}
/**
- * count callbak error times
+ * count ack fail times
*
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addCallbackErrorTimes(int num) {
+ public SortClientStateCounter addAckFailTimes(long num) {
count.getAndAdd(8, num);
return this;
}
/**
- * count ack fail times
+ * count ack succ times
*
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addAckFailTimes(int num) {
+ public SortClientStateCounter addAckSuccTimes(long num) {
count.getAndAdd(9, num);
return this;
}
/**
- * count ack succ times
+ * count request manager times
*
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addAckSuccTimes(int num) {
+ public SortClientStateCounter addRequestManagerTimes(long num) {
count.getAndAdd(10, num);
return this;
}
/**
- * count callbak done times
+ * count request manager time cost
*
- * @param num int
+ * @param num long
* @return SortClientStateCounter
*/
- public SortClientStateCounter addCallbackDoneTimes(int num) {
+ public SortClientStateCounter addRequestManagerTimeCost(long num) {
count.getAndAdd(11, num);
return this;
}
/**
- * count receive event num
+ * count request manager fail times
*
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addMsgCount(int num) {
+ public SortClientStateCounter addRequestManagerFailTimes(long num) {
count.getAndAdd(12, num);
return this;
}
@@ -217,7 +217,7 @@ public class SortClientStateCounter {
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addManagerConfChangedTimes(int num) {
+ public SortClientStateCounter addManagerConfChangedTimes(long num) {
count.getAndAdd(13, num);
return this;
}
@@ -228,7 +228,7 @@ public class SortClientStateCounter {
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addRequestManagerCommonErrorTimes(int num) {
+ public SortClientStateCounter addRequestManagerCommonErrorTimes(long num) {
count.getAndAdd(14, num);
return this;
}
@@ -239,7 +239,7 @@ public class SortClientStateCounter {
* @param num int
* @return SortClientStateCounter
*/
- public SortClientStateCounter addRequestManagerParamErrorTimes(int num) {
+ public SortClientStateCounter addRequestManagerParamErrorTimes(long num) {
count.getAndAdd(15, num);
return this;
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/StatManager.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/StatManager.java
index f771be3..b4ca542 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/StatManager.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/StatManager.java
@@ -77,45 +77,46 @@ public class StatManager implements Cleanable {
}
/**
- * use for sortId
+ * use for sortTaskId
*
- * @param sortId String
- * @return ReadApiStateCounter
+ * @param sortTaskId String
+ * @return {@link SortClientStateCounter}
*/
- public SortClientStateCounter getStatistics(String sortId) {
- String key = makeKey(sortId, defaultCluster, defaultTopic,
String.valueOf(defaultPartitionId));
+ public SortClientStateCounter getStatistics(String sortTaskId) {
+ String key = makeKey(sortTaskId, defaultCluster, defaultTopic,
String.valueOf(defaultPartitionId));
return READAPISTATE
.computeIfAbsent(key,
- k -> new SortClientStateCounter(sortId,
defaultCluster, defaultTopic, defaultPartitionId));
+ k -> new SortClientStateCounter(sortTaskId,
defaultCluster, defaultTopic, defaultPartitionId));
}
/**
* use for pulsar type
*
- * @param sortId String
+ * @param sortTaskId String
* @param clusterId String
* @param topic String
- * @return ReadApiStateCounter
+ * @return {@link SortClientStateCounter}
*/
- public SortClientStateCounter getStatistics(String sortId, String
clusterId, String topic) {
- String key = makeKey(sortId, clusterId, topic,
String.valueOf(defaultPartitionId));
+ public SortClientStateCounter getStatistics(String sortTaskId, String
clusterId, String topic) {
+ String key = makeKey(sortTaskId, clusterId, topic,
String.valueOf(defaultPartitionId));
return READAPISTATE
- .computeIfAbsent(key, k -> new SortClientStateCounter(sortId,
clusterId, topic, defaultPartitionId));
+ .computeIfAbsent(key,
+ k -> new SortClientStateCounter(sortTaskId, clusterId,
topic, defaultPartitionId));
}
/**
* use for
*
- * @param sortId String
+ * @param sortTaskId String
* @param clusterId String
* @param topic String
* @param partitionId int
- * @return ReadApiStateCounter
+ * @return {@link SortClientStateCounter}
*/
- public SortClientStateCounter getStatistics(String sortId, String
clusterId, String topic, int partitionId) {
- String key = makeKey(sortId, clusterId, topic);
+ public SortClientStateCounter getStatistics(String sortTaskId, String
clusterId, String topic, int partitionId) {
+ String key = makeKey(sortTaskId, clusterId, topic);
return READAPISTATE
- .computeIfAbsent(key, k -> new SortClientStateCounter(sortId,
clusterId, topic, partitionId));
+ .computeIfAbsent(key, k -> new
SortClientStateCounter(sortTaskId, clusterId, topic, partitionId));
}
private class ProcessStatThread extends PeriodicTask {
@@ -130,7 +131,7 @@ public class StatManager implements Cleanable {
String monitorName = SortClientConfig.MONITOR_NAME;
for (SortClientStateCounter offsetCounter :
READAPISTATE.values()) {
SortClientStateCounter counter = offsetCounter.reset();
- String[] keys = new String[]{counter.sortId,
config.getLocalIp(), counter.clusterId,
+ String[] keys = new String[]{counter.sortTaskId,
config.getLocalIp(), counter.cacheClusterId,
counter.topic,
String.valueOf(counter.partitionId)};
if (reporter != null) {
logger.debug("report statistics:{} {}",
Arrays.toString(keys),
