Fix a bug where an incorrect executor used for DNS refresh query (#6092)

Motivation:
Each `CacheEntry` should be bound to a specific `RefreshingAddressResolver`, and the associated DNS refresh query must be executed using its own executor. However, there was a bug where the refresh query could be mistakenly executed by the executor of a different resolver accessing the same `CacheEntry`.

Modifications:
- Ensured that the DNS refresh query is always executed by the correct `RefreshingAddressResolver`'s executor.

Result:
- DNS refresh query is executed by the correct executor.
- Close #5891 #6003

core/src/main/java/com/linecorp/armeria/client/RefreshingAddressResolver.java

@@ -161,7 +161,7 @@ private CompletableFuture<CacheEntry> sendQueries(List<DnsQuestion> questions, S
         return resolver.resolve(questions, hostname).handle((records, cause) -> {
             if (cause != null) {
                 cause = Exceptions.peel(cause);
-                return new CacheEntry(hostname, null, questions, creationTimeNanos, cause);
+                return new CacheEntry(executor(), hostname, null, questions, creationTimeNanos, cause);
             }
 
             InetAddress inetAddress = null;
@@ -175,17 +175,18 @@ private CompletableFuture<CacheEntry> sendQueries(List<DnsQuestion> questions, S
                     break;
                 } catch (UnknownHostException e) {
                     // Should never reach here because we already validated it in extractAddressBytes.
-                    return new CacheEntry(hostname, null, questions, creationTimeNanos,
+                    return new CacheEntry(executor(), hostname, null, questions, creationTimeNanos,
                                           new IllegalArgumentException("Invalid address: " + hostname, e));
                 }
             }
 
             if (inetAddress == null) {
-                return new CacheEntry(hostname, null, questions, creationTimeNanos, new UnknownHostException(
-                        "failed to receive DNS records for " + hostname));
+                return new CacheEntry(executor(), hostname, null, questions, creationTimeNanos,
+                                      new UnknownHostException(
+                                              "failed to receive DNS records for " + hostname));
             }
 
-            return new CacheEntry(hostname, inetAddress, questions, creationTimeNanos, null);
+            return new CacheEntry(executor(), hostname, inetAddress, questions, creationTimeNanos, null);
         });
     }
 
@@ -220,8 +221,7 @@ public void onRemoval(DnsQuestion question, @Nullable List<DnsRecord> records,
         final CacheEntry entry = addressResolverCache.getIfPresent(hostname);
         if (entry != null) {
             if (entry.refreshable()) {
-                // onRemoval is invoked by the executor of 'dnsResolverCache'.
-                executor().execute(entry::refresh);
+                entry.refresh();
             } else {
                 // Remove the old CacheEntry.
                 addressResolverCache.invalidate(hostname);
@@ -263,6 +263,7 @@ public void close() {
 
     final class CacheEntry {
 
+        private final EventExecutor executor;
         private final String hostname;
         @Nullable
         private final InetAddress address;
@@ -280,8 +281,10 @@ final class CacheEntry {
         private ScheduledFuture<?> retryFuture;
         private int numAttemptsSoFar = 1;
 
-        CacheEntry(String hostname, @Nullable InetAddress address, List<DnsQuestion> questions,
-                   @Nullable Long originalCreationTimeNanos, @Nullable Throwable cause) {
+        CacheEntry(EventExecutor executor, String hostname, @Nullable InetAddress address,
+                   List<DnsQuestion> questions, @Nullable Long originalCreationTimeNanos,
+                   @Nullable Throwable cause) {
+            this.executor = executor;
             this.hostname = hostname;
             this.address = address;
             this.questions = questions;
@@ -302,8 +305,8 @@ final class CacheEntry {
                 cacheable = !DnsUtil.isDnsQueryTimedOut(unknownHostException.getCause());
 
                 if (cacheable) {
-                    negativeCacheFuture = executor().schedule(() -> addressResolverCache.invalidate(hostname),
-                                                              negativeTtl, TimeUnit.SECONDS);
+                    negativeCacheFuture = executor.schedule(() -> addressResolverCache.invalidate(hostname),
+                                                            negativeTtl, TimeUnit.SECONDS);
                 }
             }
             this.cacheable = cacheable;
@@ -330,7 +333,7 @@ Throwable cause() {
         }
 
         void clear() {
-            executor().execute(() -> {
+            executor.execute(() -> {
                 if (retryFuture != null) {
                     retryFuture.cancel(false);
                 }
@@ -341,6 +344,14 @@ void clear() {
         }
 
         void refresh() {
+            if (executor.inEventLoop()) {
+                refresh0();
+            } else {
+                executor.execute(this::refresh0);
+            }
+        }
+
+        void refresh0() {
             if (resolverClosed) {
                 return;
             }
@@ -354,10 +365,10 @@ void refresh() {
             final String hostname = address.getHostName();
             // 'sendQueries()' always successfully completes.
             sendQueries(questions, hostname, originalCreationTimeNanos).thenAccept(entry -> {
-                if (executor().inEventLoop()) {
+                if (executor.inEventLoop()) {
                     maybeUpdate(hostname, entry);
                 } else {
-                    executor().execute(() -> maybeUpdate(hostname, entry));
+                    executor.execute(() -> maybeUpdate(hostname, entry));
                 }
             });
         }
@@ -381,7 +392,7 @@ private Object maybeUpdate(String hostname, CacheEntry entry) {
                     if (retryFuture != null) {
                         retryFuture.cancel(false);
                     }
-                    this.retryFuture = executor().schedule(() -> {
+                    this.retryFuture = executor.schedule(() -> {
                         if (refreshable()) {
                             refresh();
                         } else {
@@ -440,6 +451,7 @@ long originalCreationTimeNanos() {
         @Override
         public String toString() {
             return MoreObjects.toStringHelper(this).omitNullValues()
+                              .add("executor", executor)
                               .add("hostname", hostname)
                               .add("address", address)
                               .add("questions", questions)