database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[fix](load) acquire latest token instead of oldest token in TokenManager (#34424)

Browse Source 
* [fix](load) acquire latest token instead of oldest token

* fixup
This commit is contained in:
Kaijie Chen
2024-05-06 20:19:01 +08:00
committed by yiguolei
parent aa156f0781
commit b7b843d944
1 changed files with 9 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/TokenManager.java
View File

@ -44,6 +44,7 @@ public class TokenManager {
private int thriftTimeoutMs = 300 * 1000;
private EvictingQueue<String> tokenQueue;
private String latestToken;
private ScheduledExecutorService tokenGenerator;
public TokenManager() {
@ -52,20 +53,25 @@ public class TokenManager {
public void start() {
this.tokenQueue = EvictingQueue.create(Config.token_queue_size);
// init one token to avoid async issue.
this.tokenQueue.offer(generateNewToken());
this.addNewToken(generateNewToken());
this.tokenGenerator = Executors.newScheduledThreadPool(1,
new CustomThreadFactory("token-generator"));
this.tokenGenerator.scheduleAtFixedRate(() -> tokenQueue.offer(generateNewToken()), 0,
this.tokenGenerator.scheduleAtFixedRate(() -> this.addNewToken(generateNewToken()), 0,
Config.token_generate_period_hour, TimeUnit.HOURS);
}
private void addNewToken(String token) {
tokenQueue.offer(token);
latestToken = token;
}
private String generateNewToken() {
return UUID.randomUUID().toString();
}
public String acquireToken() throws UserException {
if (Env.getCurrentEnv().isMaster() || FeConstants.runningUnitTest) {
return tokenQueue.peek();
return latestToken;
} else {
try {
return acquireTokenFromMaster();