Clear listeners when `AddressResolverGroups` close. (#10011) (#10012)

Motivation:

When an `AddressResolverGroup` closes, it can leave unwanted listeners attached to its `EventExecutor`'s termination future.

Modifications:

- Keep track of listeners attached to termination futures
- Clear listeners if the `AddressResolverGroup` closes before its associated executor(s)

Result:

Unwanted listeners no longer remain in memory after an `AddressResolverGroup` closes before its associated executor(s).

resolver/src/main/java/io/netty/resolver/AddressResolverGroup.java

@@ -19,7 +19,9 @@ package io.netty.resolver;
 import static java.util.Objects.requireNonNull;
 
 import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
+import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
@@ -41,13 +43,16 @@ public abstract class AddressResolverGroup<T extends SocketAddress> implements C
      */
     private final Map<EventExecutor, AddressResolver<T>> resolvers = new IdentityHashMap<>();
 
+    private final Map<EventExecutor, GenericFutureListener<Future<Object>>> executorTerminationListeners =
+            new IdentityHashMap<>();
+
     protected AddressResolverGroup() { }
 
     /**
      * Returns the {@link AddressResolver} associated with the specified {@link EventExecutor}. If there's no associated
-     * resolved found, this method creates and returns a new resolver instance created by
+     * resolver found, this method creates and returns a new resolver instance created by
      * {@link #newResolver(EventExecutor)} so that the new resolver is reused on another
-     * {@link #getResolver(EventExecutor)} call with the same {@link EventExecutor}.
+     * {@code #getResolver(EventExecutor)} call with the same {@link EventExecutor}.
      */
     public AddressResolver<T> getResolver(final EventExecutor executor) {
         requireNonNull(executor, "executor");
@@ -68,12 +73,15 @@ public abstract class AddressResolverGroup<T extends SocketAddress> implements C
                 }
 
                 resolvers.put(executor, newResolver);
-                executor.terminationFuture().addListener((FutureListener<Object>) future -> {
+                FutureListener<Object> terminationListener = future -> {
                     synchronized (resolvers) {
                         resolvers.remove(executor);
+                        executorTerminationListeners.remove(executor);
                     }
                     newResolver.close();
-                });
+                };
+                executorTerminationListeners.put(executor, terminationListener);
+                executor.terminationFuture().addListener(terminationListener);
 
                 r = newResolver;
             }
@@ -97,9 +105,17 @@ public abstract class AddressResolverGroup<T extends SocketAddress> implements C
         synchronized (resolvers) {
             rArray = (AddressResolver<T>[]) resolvers.values().toArray(new AddressResolver[0]);
             resolvers.clear();
+
+            for (final Map.Entry<EventExecutor, GenericFutureListener<Future<Object>>> entry :
+                    executorTerminationListeners.entrySet()) {
+
+                entry.getKey().terminationFuture().removeListener(entry.getValue());
+            }
+
+            executorTerminationListeners.clear();
         }
 
-        for (AddressResolver<T> r: rArray) {
+        for (final AddressResolver<T> r: rArray) {
             try {
                 r.close();
             } catch (Throwable t) {