Repository: netty/netty
Issue: HashedWheelTimer task scheduling behavior changed in 4.1.85

### Expected behavior

The expectation is that task scheduling behavior of HashedWheelTimer doesn't change significantly between Netty 4.1.x releases. 

### Actual behavior

The HashedWheelTimer behavior changed in some way that makes multiple integration tests to fail in Apache Pulsar. The only change in HashedWheelTimer in 4.1.85.Final is the PR #12888 . 

### Steps to reproduce

There are steps to reproduce by running a specific test in Apache Pulsar project. The instructions are in a PR in the Apache Pulsar repo:
https://github.com/apache/pulsar/pull/18599#issuecomment-1327794984

### Netty version

4.1.85.Final

### JVM version (e.g. `java -version`)

```
openjdk version "17.0.5" 2022-10-18
OpenJDK Runtime Environment Temurin-17.0.5+8 (build 17.0.5+8)
OpenJDK 64-Bit Server VM Temurin-17.0.5+8 (build 17.0.5+8, mixed mode, sharing)
```

### OS version (e.g. `uname -a`)

Linux x86_64

Issue comments:
@chrisvest I investigated the issue and didn't find a behavior change in HashedWheelTimer. I used HashedWheelTimerTest and made several variations of testExecutionOnTime. 

There might be some subtle change that causes the issue, and we'll just have to deal with that in Pulsar.
@lhotari After reading your comment, I reviewed my code and realised I made a mistake in #12888.
My optimisation based on the assumption that the tasks in a bucket are originally in the order of execution time, which are in fact not.
So breaking the loop at `execRound > currRound` may cause latter tasks not respond in time.
Sorry for making that trouble. I already made a PR #13021 to revert it.  @lhotari @chrisvest 
> @lhotari After reading your comment, I reviewed my code and realised I made a mistake in #12888.
> My optimisation based on the assumption that the tasks in a bucket are originally in the order of execution time, which are in fact not.
> So breaking the loop at `execRound > currRound` may cause latter tasks not respond in time.

@needmorecode Thanks for the quick confirmation and investigation. Your explanation makes sense. I missed that case when I was trying to add a unit test that would prove an issue.

Pull request review comments:

Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/HashedWheelTimer.java b/common/src/main/java/io/netty/util/HashedWheelTimer.java
index 126f9998104..8db3ea747e1 100644
--- a/common/src/main/java/io/netty/util/HashedWheelTimer.java
+++ b/common/src/main/java/io/netty/util/HashedWheelTimer.java
@@ -116,7 +116,6 @@ public class HashedWheelTimer implements Timer {
     private final AtomicLong pendingTimeouts = new AtomicLong(0);
     private final long maxPendingTimeouts;
     private final Executor taskExecutor;
-    private long currRound;
 
     private volatile long startTime;
 
@@ -497,14 +496,11 @@ public void run() {
                 final long deadline = waitForNextTick();
                 if (deadline > 0) {
                     int idx = (int) (tick & mask);
-                    if (idx == 0 && tick > 0) {
-                        currRound ++;
-                    }
                     processCancelledTasks();
                     HashedWheelBucket bucket =
                             wheel[idx];
                     transferTimeoutsToBuckets();
-                    bucket.expireTimeouts(deadline, currRound);
+                    bucket.expireTimeouts(deadline);
                     tick++;
                 }
             } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
@@ -540,7 +536,7 @@ private void transferTimeoutsToBuckets() {
                 }
 
                 long calculated = timeout.deadline / tickDuration;
-                timeout.execRound = calculated / wheel.length;
+                timeout.remainingRounds = (calculated - tick) / wheel.length;
 
                 final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
                 int stopIndex = (int) (ticks & mask);
@@ -630,9 +626,9 @@ private static final class HashedWheelTimeout implements Timeout, Runnable {
         @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
         private volatile int state = ST_INIT;
 
-        // execRound will be calculated and set by Worker.transferTimeoutsToBuckets() before the
+        // remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the
         // HashedWheelTimeout will be added to the correct HashedWheelBucket.
-        long execRound;
+        long remainingRounds;
 
         // This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list.
         // As only the workerThread will act on it there is no need for synchronization / volatile.
@@ -782,13 +778,13 @@ public void addTimeout(HashedWheelTimeout timeout) {
         /**
          * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
          */
-        public void expireTimeouts(long deadline, long currRound) {
+        public void expireTimeouts(long deadline) {
             HashedWheelTimeout timeout = head;
 
             // process all timeouts
             while (timeout != null) {
                 HashedWheelTimeout next = timeout.next;
-                if (timeout.execRound <= currRound) {
+                if (timeout.remainingRounds <= 0) {
                     next = remove(timeout);
                     if (timeout.deadline <= deadline) {
                         timeout.expire();
@@ -800,7 +796,7 @@ public void expireTimeouts(long deadline, long currRound) {
                 } else if (timeout.isCancelled()) {
                     next = remove(timeout);
                 } else {
-                    break;
+                    timeout.remainingRounds --;
                 }
                 timeout = next;
             }


Test patch: