more cleanup

Author: rahul2393

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManager.java

@@ -24,11 +24,11 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -104,20 +104,6 @@ static final class EndpointState {
   private final Map<String, EndpointState> endpoints = new ConcurrentHashMap<>();
   private final Set<String> transientFailureEvictedAddresses = ConcurrentHashMap.newKeySet();
 
-  /**
-   * Active addresses reported by each ChannelFinder, keyed by database id.
-   *
-   * <p>ChannelFinder instances are held via SoftReference in KeyAwareChannel, so this map uses a
-   * stable database-id key instead of a strong ChannelFinder reference. KeyAwareChannel unregisters
-   * stale entries when a finder is cleared.
-   *
-   * <p>All reads and writes to this map, and all updates to {@link
-   * #transientFailureEvictedAddresses}, are synchronized on {@link #activeAddressLock}.
-   */
-  private final Map<String, Set<String>> activeAddressesPerFinder = new ConcurrentHashMap<>();
-
-  private final Object activeAddressLock = new Object();
-
   private final ScheduledExecutorService scheduler;
   private final AtomicBoolean isShutdown = new AtomicBoolean(false);
   private final long probeIntervalSeconds;
@@ -164,52 +150,23 @@ static final class EndpointState {
             TimeUnit.SECONDS);
   }
 
-  /**
-   * Ensures an endpoint state exists for the given address.
-   *
-   * <p>This is only called from {@link #updateActiveAddresses} under {@link #activeAddressLock} to
-   * guarantee that newly created endpoints are registered as active before any stale-eviction check
-   * can see them. Background creation tasks are scheduled by the caller after {@code
-   * computeIfAbsent} returns, so the entry is visible in the map before the scheduler thread checks
-   * it.
-   *
-   * @return true if a new endpoint state was created (caller should schedule background creation)
-   */
-  private boolean ensureEndpointExists(String address) {
-    if (isShutdown.get() || address == null || address.isEmpty()) {
-      return false;
-    }
-    // Don't manage the default endpoint.
-    if (defaultEndpointAddress.equals(address)) {
-      return false;
-    }
-
-    boolean[] created = {false};
-    endpoints.computeIfAbsent(
-        address,
-        addr -> {
-          logger.log(Level.FINE, "Creating endpoint state for address: {0}", addr);
-          created[0] = true;
-          return new EndpointState(addr, clock.instant());
-        });
-    return created[0];
-  }
-
-  private void retainTransientFailureEvictionMarkers(Set<String> activeAddresses) {
-    synchronized (activeAddressLock) {
-      transientFailureEvictedAddresses.retainAll(activeAddresses);
-    }
-  }
-
   private void markTransientFailureEvicted(String address) {
-    synchronized (activeAddressLock) {
-      transientFailureEvictedAddresses.add(address);
-    }
+    transientFailureEvictedAddresses.add(address);
   }
 
   private void clearTransientFailureEvictionMarker(String address) {
-    synchronized (activeAddressLock) {
-      transientFailureEvictedAddresses.remove(address);
+    transientFailureEvictedAddresses.remove(address);
+  }
+
+  private void submitLifecycleTask(String description, Runnable task) {
+    try {
+      scheduler.submit(task);
+    } catch (RejectedExecutionException e) {
+      logger.log(
+          Level.FINE,
+          String.format(
+              "Skipping lifecycle task '%s': lifecycle manager is shutting down", description),
+          e);
     }
   }
 
@@ -234,102 +191,7 @@ void recordRealTraffic(String address) {
             });
     state.lastRealTrafficAt = now;
     if (created[0]) {
-      scheduler.submit(() -> startProbing(address));
-    }
-  }
-
-  /**
-   * Atomically ensures endpoints exist for all active addresses and evicts any managed endpoints
-   * that are no longer referenced by any finder. This handles the case where a tablet's server
-   * address changes (e.g. from server1:15000 to server2:15000) — the old endpoint is shut down
-   * promptly instead of lingering until idle eviction.
-   *
-   * <p>Both endpoint creation and stale-eviction are performed under the same lock to prevent a
-   * race condition where a newly created endpoint could be evicted by a concurrent call from
-   * another finder before it is registered as active.
-   *
-   * @param finderKey stable identifier of the ChannelFinder reporting its active addresses
-   * @param activeAddresses server addresses currently referenced by tablets in this finder
-   */
-  void updateActiveAddresses(String finderKey, Set<String> activeAddresses) {
-    if (isShutdown.get() || finderKey == null || finderKey.isEmpty()) {
-      return;
-    }
-    List<String> newlyCreated = new ArrayList<>();
-    synchronized (activeAddressLock) {
-      // Ensure endpoints exist for all active addresses while holding the lock.
-      // This guarantees the addresses are in the endpoints map before we compute stale entries.
-      for (String address : activeAddresses) {
-        if (ensureEndpointExists(address)) {
-          newlyCreated.add(address);
-        }
-      }
-
-      activeAddressesPerFinder.put(finderKey, activeAddresses);
-
-      // Compute the union of all active addresses across all finders.
-      Set<String> allActive = new HashSet<>();
-      for (Set<String> addresses : activeAddressesPerFinder.values()) {
-        allActive.addAll(addresses);
-      }
-      retainTransientFailureEvictionMarkers(allActive);
-
-      // Evict managed endpoints not referenced by any finder.
-      List<String> stale = new ArrayList<>();
-      for (String address : endpoints.keySet()) {
-        if (!allActive.contains(address)) {
-          stale.add(address);
-        }
-      }
-
-      for (String address : stale) {
-        logger.log(
-            Level.FINE, "Evicting stale endpoint {0}: no longer referenced by any tablet", address);
-        evictEndpoint(address);
-      }
-    }
-
-    // Schedule background creation tasks AFTER computeIfAbsent has returned and the entries
-    // are visible to other threads. Submitting from inside computeIfAbsent creates a race
-    // where the scheduler thread can run before the entry is published in the map.
-    for (String address : newlyCreated) {
-      scheduler.submit(() -> createAndStartProbing(address));
-    }
-  }
-
-  /**
-   * Unregisters a finder and evicts any managed endpoints that are no longer referenced by the
-   * remaining finders.
-   */
-  void unregisterFinder(String finderKey) {
-    if (isShutdown.get() || finderKey == null || finderKey.isEmpty()) {
-      return;
-    }
-    synchronized (activeAddressLock) {
-      if (activeAddressesPerFinder.remove(finderKey) == null) {
-        return;
-      }
-
-      Set<String> allActive = new HashSet<>();
-      for (Set<String> addresses : activeAddressesPerFinder.values()) {
-        allActive.addAll(addresses);
-      }
-      retainTransientFailureEvictionMarkers(allActive);
-
-      List<String> stale = new ArrayList<>();
-      for (String address : endpoints.keySet()) {
-        if (!allActive.contains(address)) {
-          stale.add(address);
-        }
-      }
-
-      for (String address : stale) {
-        logger.log(
-            Level.FINE,
-            "Evicting stale endpoint {0}: finder {1} was unregistered",
-            new Object[] {address, finderKey});
-        evictEndpoint(address);
-      }
+      submitLifecycleTask("startProbing", () -> startProbing(address));
     }
   }
 
@@ -567,7 +429,7 @@ void requestEndpointRecreation(String address) {
     EndpointState state = new EndpointState(address, clock.instant());
     if (endpoints.putIfAbsent(address, state) == null) {
       // Schedule after putIfAbsent returns so the entry is visible to the scheduler thread.
-      scheduler.submit(() -> createAndStartProbing(address));
+      submitLifecycleTask("createAndStartProbing", () -> createAndStartProbing(address));
     }
   }
 

java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/EndpointLifecycleManagerTest.java

@@ -26,11 +26,7 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneId;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.LockSupport;
 import java.util.function.BooleanSupplier;
 import org.junit.After;
@@ -43,26 +39,17 @@ public class EndpointLifecycleManagerTest {
 
   private EndpointLifecycleManager manager;
 
-  /** Counter for generating unique finder keys in tests. */
-  private static final AtomicLong TEST_FINDER_ID = new AtomicLong(1000);
-
   @After
   public void tearDown() {
     if (manager != null) {
       manager.shutdown();
     }
   }
 
-  /**
-   * Registers addresses with the lifecycle manager via updateActiveAddresses, which atomically
-   * creates endpoints and registers them as active. This mirrors how ChannelFinder.update() works.
-   */
-  private static String registerAddresses(EndpointLifecycleManager mgr, String... addresses) {
-    String finderId = "finder-" + TEST_FINDER_ID.incrementAndGet();
-    Set<String> addressSet = new HashSet<>();
-    Collections.addAll(addressSet, addresses);
-    mgr.updateActiveAddresses(finderId, addressSet);
-    return finderId;
+  private static void registerAddresses(EndpointLifecycleManager mgr, String... addresses) {
+    for (String address : addresses) {
+      mgr.requestEndpointRecreation(address);
+    }
   }
 
   private static void awaitCondition(String message, BooleanSupplier condition) {
@@ -103,10 +90,9 @@ public void duplicateRegistrationIsNoop() throws Exception {
         new EndpointLifecycleManager(
             cache, /* probeIntervalSeconds= */ 60, Duration.ofMinutes(30), Clock.systemUTC());
 
-    String finderId = registerAddresses(manager, "server1");
-    // Re-register with the same finder ID — should not create duplicate state.
-    manager.updateActiveAddresses(finderId, Collections.singleton("server1"));
-    manager.updateActiveAddresses(finderId, Collections.singleton("server1"));
+    registerAddresses(manager, "server1");
+    manager.requestEndpointRecreation("server1");
+    manager.requestEndpointRecreation("server1");
 
     assertEquals(1, manager.managedEndpointCount());
   }
@@ -118,7 +104,7 @@ public void defaultEndpointIsNotManaged() {
         new EndpointLifecycleManager(
             cache, /* probeIntervalSeconds= */ 60, Duration.ofMinutes(30), Clock.systemUTC());
 
-    registerAddresses(manager, "default");
+    manager.requestEndpointRecreation("default");
 
     assertFalse(manager.isManaged("default"));
     assertEquals(0, manager.managedEndpointCount());
@@ -155,8 +141,6 @@ public void realRoutedTrafficUpdatesLastRealTrafficAt() throws Exception {
         new EndpointLifecycleManager(
             cache, /* probeIntervalSeconds= */ 60, Duration.ofMinutes(30), clock);
 
-    registerAddresses(manager, "server1");
-
     Instant before = clock.instant();
     clock.advance(Duration.ofMinutes(5));
     manager.recordRealTraffic("server1");
@@ -267,28 +251,6 @@ public void transientFailureEvictionTrackedUntilEndpointReadyAgain() throws Exce
         () -> !manager.wasRecentlyEvictedTransientFailure("server1"));
   }
 
-  @Test
-  public void transientFailureEvictionMarkerRemovedWhenAddressNoLongerActive() throws Exception {
-    KeyRangeCacheTest.FakeEndpointCache cache = new KeyRangeCacheTest.FakeEndpointCache();
-    manager =
-        new EndpointLifecycleManager(
-            cache, /* probeIntervalSeconds= */ 1, Duration.ofMinutes(30), Clock.systemUTC());
-
-    String finder = registerAddresses(manager, "server1");
-    awaitCondition(
-        "endpoint should be created in background", () -> cache.getIfPresent("server1") != null);
-
-    cache.setState("server1", KeyRangeCacheTest.EndpointHealthState.TRANSIENT_FAILURE);
-    awaitCondition(
-        "endpoint should be evicted after repeated transient failures",
-        () ->
-            !manager.isManaged("server1") && manager.wasRecentlyEvictedTransientFailure("server1"));
-
-    manager.updateActiveAddresses(finder, Collections.emptySet());
-
-    assertFalse(manager.wasRecentlyEvictedTransientFailure("server1"));
-  }
-
   @Test
   public void shutdownStopsAllProbing() throws Exception {
     KeyRangeCacheTest.FakeEndpointCache cache = new KeyRangeCacheTest.FakeEndpointCache();
@@ -312,8 +274,8 @@ public void emptyOrNullAddressIsIgnored() {
         new EndpointLifecycleManager(
             cache, /* probeIntervalSeconds= */ 60, Duration.ofMinutes(30), Clock.systemUTC());
 
-    manager.updateActiveAddresses("finder-1", Collections.singleton(""));
-    manager.updateActiveAddresses("finder-2", Collections.emptySet());
+    manager.requestEndpointRecreation("");
+    manager.requestEndpointRecreation(null);
 
     assertEquals(0, manager.managedEndpointCount());
   }
@@ -331,66 +293,6 @@ public void recordRealTrafficForDefaultEndpointIsIgnored() {
     assertEquals(0, manager.managedEndpointCount());
   }
 
-  @Test
-  public void staleEndpointEvictedWhenNoLongerActive() throws Exception {
-    KeyRangeCacheTest.FakeEndpointCache cache = new KeyRangeCacheTest.FakeEndpointCache();
-    manager =
-        new EndpointLifecycleManager(
-            cache, /* probeIntervalSeconds= */ 60, Duration.ofMinutes(30), Clock.systemUTC());
-
-    // Finder 1 reports server1 and server2.
-    String finder1 = registerAddresses(manager, "server1", "server2");
-    assertEquals(2, manager.managedEndpointCount());
-
-    // Finder 1 updates: server1 is gone, only server2 remains.
-    manager.updateActiveAddresses(finder1, Collections.singleton("server2"));
-
-    // server1 should be evicted since no finder references it.
-    assertFalse(manager.isManaged("server1"));
-    assertTrue(manager.isManaged("server2"));
-    assertEquals(1, manager.managedEndpointCount());
-  }
-
-  @Test
-  public void endpointKeptIfReferencedByAnotherFinder() throws Exception {
-    KeyRangeCacheTest.FakeEndpointCache cache = new KeyRangeCacheTest.FakeEndpointCache();
-    manager =
-        new EndpointLifecycleManager(
-            cache, /* probeIntervalSeconds= */ 60, Duration.ofMinutes(30), Clock.systemUTC());
-
-    // Finder 1 reports server1.
-    String finder1 = registerAddresses(manager, "server1");
-    // Finder 2 also reports server1.
-    registerAddresses(manager, "server1");
-
-    // Finder 1 drops server1, but finder 2 still references it.
-    manager.updateActiveAddresses(finder1, Collections.emptySet());
-
-    assertTrue(manager.isManaged("server1"));
-    assertEquals(1, manager.managedEndpointCount());
-  }
-
-  @Test
-  public void unregisterFinderEvictsEndpointsNoLongerReferenced() throws Exception {
-    KeyRangeCacheTest.FakeEndpointCache cache = new KeyRangeCacheTest.FakeEndpointCache();
-    manager =
-        new EndpointLifecycleManager(
-            cache, /* probeIntervalSeconds= */ 60, Duration.ofMinutes(30), Clock.systemUTC());
-
-    String finder1 = registerAddresses(manager, "server1");
-    String finder2 = registerAddresses(manager, "server2");
-
-    manager.unregisterFinder(finder1);
-
-    assertFalse(manager.isManaged("server1"));
-    assertTrue(manager.isManaged("server2"));
-    assertEquals(1, manager.managedEndpointCount());
-
-    manager.unregisterFinder(finder2);
-
-    assertEquals(0, manager.managedEndpointCount());
-  }
-
   /** Test clock that can be advanced manually. */
   private static final class TestClock extends Clock {
     private Instant now;