CosmosNi commented on code in PR #11226:
URL: https://github.com/apache/seatunnel/pull/11226#discussion_r3497462354
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/StateStoreCheckpointIDCounter.java:
##########
@@ -57,26 +57,26 @@ public void start() throws Exception {
@Override
public CompletableFuture<Void> shutdown(PipelineStatus pipelineStatus) {
if (pipelineStatus.isEndState()) {
- checkpointIdMap.remove(key);
+ checkpointCounterStore.remove(key);
}
return CompletableFuture.completedFuture(null);
}
@Override
public long getAndIncrement() throws Exception {
- Long nextId = checkpointIdMap.compute(key, (k, v) -> v == null ? null
: v + 1);
+ long nextId = checkpointCounterStore.incrementAndGet(key);
checkNotNull(nextId);
Review Comment:
The default value of nextId is 1L, so the checkNotNull check has no effect.
Alternatively, should incrementAndGet return -1 to throw an exception? Since
the start method will initialize it, if it is null, that indicates the state
has been lost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
