This is an automated email from the ASF dual-hosted git repository.
gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new f61167ac4b [INLONG-8321][TubeMQ] Improve the precision of tube
consumer id (#8325)
f61167ac4b is described below
commit f61167ac4b33e85d654039ff95d571db5f33c67b
Author: Goson Zhang <[email protected]>
AuthorDate: Tue Jun 27 10:12:46 2023 +0800
[INLONG-8321][TubeMQ] Improve the precision of tube consumer id (#8325)
---
.../inlong/tubemq/client/consumer/BaseMessageConsumer.java | 8 +++++---
.../tubemq/client/consumer/SimpleClientBalanceConsumer.java | 9 +++++----
.../apache/inlong/tubemq/client/producer/ProducerManager.java | 9 +++++----
3 files changed, 15 insertions(+), 11 deletions(-)
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/BaseMessageConsumer.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/BaseMessageConsumer.java
index eeeaed82db..543bfb9bc7 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/BaseMessageConsumer.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/BaseMessageConsumer.java
@@ -52,6 +52,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.lang.management.ManagementFactory;
+import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -79,7 +80,8 @@ public class BaseMessageConsumer implements MessageConsumer {
private static final Logger logger =
LoggerFactory.getLogger(BaseMessageConsumer.class);
private static final int REBALANCE_QUEUE_SIZE = 5000;
- private static final AtomicInteger consumerCounter = new AtomicInteger(0);
+ private static final SecureRandom sRandom = new SecureRandom(
+ Long.toString(System.nanoTime()).getBytes());
protected final String consumerId;
protected final ConsumerConfig consumerConfig;
protected final RmtDataCache rmtDataCache;
@@ -590,8 +592,8 @@ public class BaseMessageConsumer implements MessageConsumer
{
.append(this.consumerConfig.getConsumerGroup())
.append("_").append(AddressUtils.getLocalAddress())
.append("-").append(pidName)
- .append("-").append(System.currentTimeMillis())
- .append("-").append(consumerCounter.incrementAndGet());
+ .append("-").append(System.nanoTime())
+ .append("-").append(Math.abs(sRandom.nextInt()));
if (this.isPullConsume) {
strBuffer.append("-Pull-");
} else {
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/SimpleClientBalanceConsumer.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/SimpleClientBalanceConsumer.java
index e1d73f02ce..7113585b07 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/SimpleClientBalanceConsumer.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/SimpleClientBalanceConsumer.java
@@ -52,6 +52,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.management.ManagementFactory;
+import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -79,8 +80,8 @@ public class SimpleClientBalanceConsumer implements
ClientBalanceConsumer {
private static final Logger logger =
LoggerFactory.getLogger(SimpleClientBalanceConsumer.class);
- private static final AtomicInteger consumerCounter =
- new AtomicInteger(0);
+ private static final SecureRandom sRandom = new SecureRandom(
+ Long.toString(System.nanoTime()).getBytes());
protected final String consumerId;
protected final ConsumerConfig consumerConfig;
private final InnerSessionFactory sessionFactory;
@@ -1753,8 +1754,8 @@ public class SimpleClientBalanceConsumer implements
ClientBalanceConsumer {
.append(this.consumerConfig.getConsumerGroup())
.append("_").append(AddressUtils.getLocalAddress())
.append("-").append(pidName)
- .append("-").append(System.currentTimeMillis())
- .append("-").append(consumerCounter.incrementAndGet())
+ .append("-").append(System.nanoTime())
+ .append("-").append(Math.abs(sRandom.nextInt()))
.append("-Balance-")
.append(TubeClientVersion.CONSUMER_VERSION).toString();
}
diff --git
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/producer/ProducerManager.java
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/producer/ProducerManager.java
index e7e166de97..f9e2843160 100644
---
a/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/producer/ProducerManager.java
+++
b/inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/producer/ProducerManager.java
@@ -48,6 +48,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.management.ManagementFactory;
+import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -71,8 +72,8 @@ public class ProducerManager {
private static final Logger logger =
LoggerFactory.getLogger(ProducerManager.class);
private static final int BROKER_UPDATED_TIME_AFTER_RETRY_FAIL = 2 * 60 *
60 * 1000;
- private static final AtomicInteger producerCounter =
- new AtomicInteger(0);
+ private static final SecureRandom sRandom = new SecureRandom(
+ Long.toString(System.nanoTime()).getBytes());
private final String producerId;
private final int producerAddrId;
private final TubeClientConfig tubeClientConfig;
@@ -560,8 +561,8 @@ public class ProducerManager {
return new StringBuilder(256)
.append(AddressUtils.getLocalAddress())
.append("-").append(pidName)
- .append("-").append(System.currentTimeMillis())
- .append("-").append(producerCounter.incrementAndGet())
+ .append("-").append(System.nanoTime())
+ .append("-").append(Math.abs(sRandom.nextInt()))
.append("-").append(TubeClientVersion.PRODUCER_VERSION).toString();
}
