Repository: apache/hertzbeat (language: java)

Issue:
[BUG]  pendingTimeouts may be incorrect in the HashedWheelTimer
### Is there an existing issue for this?

- [x] I have searched the existing issues

### Current Behavior

If a `Timeout` in `HashedWheelTimer` has been transferred to a bucket and is canceled before it is actually expired in the goal tick, `pendingTimeouts` may be repeatedly decremented.
If the user sets a positive `maxPendingTimeouts,` this bug means that the maximum number of tasks can be potentially increased.

### Expected Behavior

The `pendingTimeouts` in `HashedWheelTimer` can correctly reflect the number of waiting tasks.

### Steps To Reproduce

The following test code reproduces the bug.
```
final HashedWheelTimer timer = new HashedWheelTimer();
final CountDownLatch barrier = new CountDownLatch(1);
// A total of 11 timeouts with the same delay are submitted here, and they will be processed in the same tick.
timer.newTimeout(t -> {
    barrier.countDown();
    Thread.sleep(1000);
}, 200, TimeUnit.MILLISECONDS);
List<Timeout> timeouts = new ArrayList<>();
for (int i = 0; i < 10; i++) {
    timeouts.add(timer.newTimeout(createNoOpTimerTask(), 200, TimeUnit.MILLISECONDS));
}
barrier.await();
// The simulation here is that the timeout has been transferred to a bucket and is canceled before it is
// actually expired in the goal tick.
timeouts.forEach(Timeout::cancel);
Thread.sleep(2000);
assertEquals(0, timer.pendingTimeouts());
timer.stop();
```
Got the following test failures and ended up with a negative `pendingTimeouts`.
```
Expected :0
Actual   :-10
```

### Environment

```markdown
HertzBeat version(s): The latest master branch.
```

### Debug logs

_No response_

### Anything else?

_No response_


Discussion / hints:

Fix patch:
diff --git a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/timer/HashedWheelTimer.java b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/timer/HashedWheelTimer.java
index 5d6bd6f3e82..184cbf4b0d7 100644
--- a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/timer/HashedWheelTimer.java
+++ b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/timer/HashedWheelTimer.java
@@ -613,9 +613,8 @@ void remove() {
             HashedWheelBucket bucket = this.bucket;
             if (bucket != null) {
                 bucket.remove(this);
-            } else {
-                timer.pendingTimeouts.decrementAndGet();
             }
+            timer.pendingTimeouts.decrementAndGet();
         }
 
         public boolean compareAndSetState(int expected, int state) {
@@ -642,6 +641,7 @@ public void expire() {
             }
 
             try {
+                remove();
                 task.run(this);
             } catch (Throwable t) {
                 if (logger.isWarnEnabled()) {
@@ -718,7 +718,6 @@ void expireTimeouts(long deadline) {
             while (timeout != null) {
                 HashedWheelTimeout next = timeout.next;
                 if (timeout.remainingRounds <= 0) {
-                    next = remove(timeout);
                     if (timeout.deadline <= deadline) {
                         timeout.expire();
                     } else {
@@ -726,9 +725,7 @@ void expireTimeouts(long deadline) {
                         throw new IllegalStateException(String.format(
                                 "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                     }
-                } else if (timeout.isCancelled()) {
-                    next = remove(timeout);
-                } else {
+                } else if (!timeout.isCancelled()) {
                     timeout.remainingRounds--;
                 }
                 timeout = next;
@@ -761,7 +758,6 @@ public HashedWheelTimeout remove(HashedWheelTimeout timeout) {
             timeout.prev = null;
             timeout.next = null;
             timeout.bucket = null;
-            timeout.timer.pendingTimeouts.decrementAndGet();
             return next;
         }
 


Test patch: