fix(cassandra): fix session poisoning on transient failures

- Reset driver session after each transient failure in executeWithRetry()
  so retries reopen cleanly via lazy open()
- Remove redundant finally block in reconnectIfNeeded(); null session
  directly on DriverException
- Store retryBaseDelay as field, reuse in open() (removes double-read)
- One-time LOG.warn via AtomicBoolean for commitAsync() retry gap
- Tighten defaults: max_delay 60s→10s, max_retries 10→3, interval 5s→1s
- Wire retry config via HugeConfig in tests; add cross-validator tests

Author: dpol1

hugegraph-server/hugegraph-cassandra/src/main/java/org/apache/hugegraph/backend/store/cassandra/CassandraOptions.java

@@ -148,7 +148,7 @@ public static synchronized CassandraOptions instance() {
                     "exponential reconnection policy when a Cassandra host " +
                     "becomes unreachable.",
                     rangeInt(1000L, Long.MAX_VALUE),
-                    60_000L
+                    10_000L
             );
 
     public static final ConfigOption<Integer> CASSANDRA_RECONNECT_MAX_RETRIES =
@@ -159,7 +159,7 @@ public static synchronized CassandraOptions instance() {
                     "(NoHostAvailableException / OperationTimedOutException). " +
                     "Set to 0 to disable query-time retries.",
                     rangeInt(0, Integer.MAX_VALUE),
-                    10
+                    3
             );
 
     public static final ConfigOption<Long> CASSANDRA_RECONNECT_INTERVAL =
@@ -170,6 +170,6 @@ public static synchronized CassandraOptions instance() {
                     "actual wait grows with exponential backoff, capped at " +
                     "cassandra.reconnect_max_delay.",
                     rangeInt(100L, Long.MAX_VALUE),
-                    5000L
+                    1000L
             );
 }

hugegraph-server/hugegraph-cassandra/src/main/java/org/apache/hugegraph/backend/store/cassandra/CassandraSessionPool.java

@@ -20,6 +20,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.hugegraph.backend.BackendException;
 import org.apache.hugegraph.backend.store.BackendSession.AbstractBackendSession;
@@ -53,10 +54,21 @@ public class CassandraSessionPool extends BackendSessionPool {
     private static final String HEALTH_CHECK_CQL =
             "SELECT now() FROM system.local";
 
+    /**
+     * Guards the one-time JVM-wide warning about {@code commitAsync()} not
+     * being covered by query-time retries. {@link CassandraSessionPool} is
+     * instantiated once per backend store per graph, so without this guard
+     * the warning would fire many times on startup for a structural
+     * limitation that does not change between instances.
+     */
+    private static final AtomicBoolean ASYNC_RETRY_WARNING_LOGGED =
+            new AtomicBoolean(false);
+
     private Cluster cluster;
     private final String keyspace;
     private final int maxRetries;
     private final long retryInterval;
+    private final long retryBaseDelay;
     private final long retryMaxDelay;
 
     public CassandraSessionPool(HugeConfig config,
@@ -78,7 +90,14 @@ public CassandraSessionPool(HugeConfig config,
                         reconnectMax,
                         CassandraOptions.CASSANDRA_RECONNECT_BASE_DELAY.name(),
                         reconnectBase);
+        this.retryBaseDelay = reconnectBase;
         this.retryMaxDelay = reconnectMax;
+
+        if (this.maxRetries > 0 &&
+            ASYNC_RETRY_WARNING_LOGGED.compareAndSet(false, true)) {
+            LOG.warn("cassandra.reconnect_max_retries={} applies to sync commit()" +
+                     " only. commitAsync() has no retry protection.", this.maxRetries);
+        }
     }
 
     @Override
@@ -117,10 +136,8 @@ public synchronized void open() {
 
         // Reconnection policy: let driver keep retrying nodes in background
         // with exponential backoff after they go down (see issue #2740).
-        long reconnectBase = config.get(
-                CassandraOptions.CASSANDRA_RECONNECT_BASE_DELAY);
         builder.withReconnectionPolicy(
-                new ExponentialReconnectionPolicy(reconnectBase,
+                new ExponentialReconnectionPolicy(this.retryBaseDelay,
                                                   this.retryMaxDelay));
 
         // Credential options
@@ -211,7 +228,11 @@ public void commitAsync() {
             int processors = Math.min(statements.size(), 1023);
             List<ResultSetFuture> results = new ArrayList<>(processors + 1);
             for (Statement s : statements) {
-                // TODO: commitAsync is not retried (async retry semantics are complex)
+                // TODO(issue #2740): commitAsync() bypasses executeWithRetry().
+                // During a Cassandra restart, async writes may fail with
+                // NoHostAvailableException even when maxRetries > 0. Callers
+                // must handle CompletableFuture failures. A follow-up will
+                // wrap each future with retry semantics.
                 ResultSetFuture future = this.session.executeAsync(s);
                 results.add(future);
 
@@ -253,13 +274,19 @@ public ResultSet execute(String statement, Object... args) {
          * reconnection policy, so once Cassandra comes back online, a
          * subsequent attempt here will succeed without restarting the server.
          *
+         * <p>If the driver session has been discarded (e.g. by
+         * {@link #reconnectIfNeeded()} after a failed health-check) it is
+         * lazily reopened at the start of each attempt. After a transient
+         * failure the session is {@linkplain #reset() reset} so the next
+         * iteration gets a fresh driver session.
+         *
          * <p><b>Blocking note:</b> retries block the calling thread via
          * {@link Thread#sleep(long)}. Worst-case a single call blocks for
          * {@code maxRetries * retryMaxDelay} ms. Under high-throughput
          * workloads concurrent threads may pile up in {@code sleep()} during
          * a Cassandra outage. For such deployments lower
-         * {@code cassandra.reconnect_max_retries} (default 10) and
-         * {@code cassandra.reconnect_max_delay} (default 60000ms) so the
+         * {@code cassandra.reconnect_max_retries} (default 3) and
+         * {@code cassandra.reconnect_max_delay} (default 10000ms) so the
          * request fails fast and pressure is released back to the caller.
          */
         private ResultSet executeWithRetry(Statement statement) {
@@ -269,9 +296,18 @@ private ResultSet executeWithRetry(Statement statement) {
             DriverException lastError = null;
             for (int attempt = 0; attempt <= retries; attempt++) {
                 try {
+                    if (this.session == null) {
+                        // Lazy reopen: may itself throw NHAE while
+                        // Cassandra is still unreachable; the catch below
+                        // treats that as a transient failure.
+                        this.open();
+                    }
                     return this.session.execute(statement);
                 } catch (NoHostAvailableException | OperationTimedOutException e) {
                     lastError = e;
+                    // Discard the (possibly broken) driver session so the
+                    // next iteration reopens cleanly.
+                    this.reset();
                     if (attempt >= retries) {
                         break;
                     }
@@ -359,9 +395,10 @@ public boolean hasChanges() {
          * Periodic liveness probe invoked by {@link BackendSessionPool} to
          * recover thread-local sessions after Cassandra has been restarted.
          * Reopens the driver session if it was closed and pings the cluster
-         * with a lightweight query. Any failure here is swallowed so the
-         * caller can still issue the real query, which will drive retries via
-         * {@link #executeWithRetry(Statement)}.
+         * with a lightweight query. On failure the session is discarded via
+         * {@link #reset()} so the next call to
+         * {@link #executeWithRetry(Statement)} reopens it; any exception
+         * here is swallowed so the caller can still issue the real query.
          */
         @Override
         public void reconnectIfNeeded() {
@@ -377,15 +414,9 @@ public void reconnectIfNeeded() {
                     this.session.execute(new SimpleStatement(HEALTH_CHECK_CQL));
                 }
             } catch (DriverException e) {
-                LOG.debug("Cassandra health-check failed, " +
-                          "will retry on next query: {}", e.getMessage());
-            } finally {
-                // Keep opened flag consistent with session: if tryOpen()
-                // failed to reopen, clear opened so the next execute() does
-                // not NPE before executeWithRetry() can intercept.
-                if (this.session == null) {
-                    this.opened = false;
-                }
+                LOG.debug("Cassandra health-check failed, resetting session: {}",
+                          e.getMessage());
+                this.session = null;
             }
         }
 

hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/cassandra/CassandraTest.java

@@ -205,11 +205,11 @@ public void testReconnectOptionsHaveSensibleDefaults() {
         // HugeGraph keeps running when Cassandra restarts (issue #2740).
         Assert.assertEquals(1000L, (long) CassandraOptions
                 .CASSANDRA_RECONNECT_BASE_DELAY.defaultValue());
-        Assert.assertEquals(60_000L, (long) CassandraOptions
+        Assert.assertEquals(10_000L, (long) CassandraOptions
                 .CASSANDRA_RECONNECT_MAX_DELAY.defaultValue());
-        Assert.assertEquals(10, (int) CassandraOptions
+        Assert.assertEquals(3, (int) CassandraOptions
                 .CASSANDRA_RECONNECT_MAX_RETRIES.defaultValue());
-        Assert.assertEquals(5000L, (long) CassandraOptions
+        Assert.assertEquals(1000L, (long) CassandraOptions
                 .CASSANDRA_RECONNECT_INTERVAL.defaultValue());
     }
 
@@ -251,13 +251,22 @@ public void testReconnectRetriesCanBeDisabled() {
 
     @Test
     public void testExecuteWithRetrySucceedsAfterTransientFailures() {
+        // Configure retry knobs via config so the pool reads them through
+        // the normal path (no Whitebox overrides on retry fields). Keep the
+        // values within the validators' lower bounds (base >= 100, max >=
+        // base, interval >= 100).
         Configuration conf = new PropertiesConfiguration();
+        conf.setProperty(
+                CassandraOptions.CASSANDRA_RECONNECT_BASE_DELAY.name(), 100L);
+        conf.setProperty(
+                CassandraOptions.CASSANDRA_RECONNECT_MAX_DELAY.name(), 1000L);
+        conf.setProperty(
+                CassandraOptions.CASSANDRA_RECONNECT_MAX_RETRIES.name(), 3);
+        conf.setProperty(
+                CassandraOptions.CASSANDRA_RECONNECT_INTERVAL.name(), 100L);
         HugeConfig config = new HugeConfig(conf);
         CassandraSessionPool pool = new CassandraSessionPool(config,
                                                              "ks", "store");
-        Whitebox.setInternalState(pool, "maxRetries", 3);
-        Whitebox.setInternalState(pool, "retryInterval", 1L);
-        Whitebox.setInternalState(pool, "retryMaxDelay", 10L);
 
         com.datastax.driver.core.Session driverSession = Mockito.mock(
                 com.datastax.driver.core.Session.class);
@@ -269,6 +278,17 @@ public void testExecuteWithRetrySucceedsAfterTransientFailures() {
                .thenThrow(transientFailure)
                .thenReturn(rs);
 
+        // executeWithRetry now resets the driver session on transient
+        // failures, so the next iteration calls cluster().connect(keyspace)
+        // to obtain a fresh one. Install a mocked Cluster that hands back
+        // the same driverSession for each reconnect.
+        com.datastax.driver.core.Cluster mockCluster = Mockito.mock(
+                com.datastax.driver.core.Cluster.class);
+        Mockito.when(mockCluster.isClosed()).thenReturn(false);
+        Mockito.when(mockCluster.connect(Mockito.anyString()))
+               .thenReturn(driverSession);
+        Whitebox.setInternalState(pool, "cluster", mockCluster);
+
         CassandraSessionPool.Session session = pool.new Session();
         Whitebox.setInternalState(session, "session", driverSession);
 
@@ -279,18 +299,38 @@ public void testExecuteWithRetrySucceedsAfterTransientFailures() {
     }
 
     @Test
-    public void testReconnectOptionsExposeExpectedKeys() {
-        Assert.assertEquals("cassandra.reconnect_base_delay",
-                            CassandraOptions.CASSANDRA_RECONNECT_BASE_DELAY
-                                            .name());
-        Assert.assertEquals("cassandra.reconnect_max_delay",
-                            CassandraOptions.CASSANDRA_RECONNECT_MAX_DELAY
-                                            .name());
-        Assert.assertEquals("cassandra.reconnect_max_retries",
-                            CassandraOptions.CASSANDRA_RECONNECT_MAX_RETRIES
-                                            .name());
-        Assert.assertEquals("cassandra.reconnect_interval",
-                            CassandraOptions.CASSANDRA_RECONNECT_INTERVAL
-                                            .name());
+    public void testReconnectBaseDelayBelowMinimumRejected() {
+        // The validator on CASSANDRA_RECONNECT_BASE_DELAY is
+        // rangeInt(100L, Long.MAX_VALUE); values below 100 must be rejected
+        // at parse time. Setting the property as a String forces HugeConfig
+        // to run parseConvert() which invokes the range check.
+        Configuration conf = new PropertiesConfiguration();
+        Assert.assertThrows(Exception.class, () -> {
+            conf.setProperty(
+                    CassandraOptions.CASSANDRA_RECONNECT_BASE_DELAY.name(),
+                    "50");
+            new HugeConfig(conf);
+        });
+    }
+
+    @Test
+    public void testReconnectMaxDelayLessThanBaseRejected() {
+        // Both values must pass their individual range validators with margin
+        // (base >= 100, max >= 1000), so the only thing that can throw is the
+        // E.checkArgument(max >= base) cross-check inside the pool ctor. Set
+        // all four reconnect properties explicitly so the test does not depend
+        // on default values changing in CassandraOptions.
+        Configuration conf = new PropertiesConfiguration();
+        conf.setProperty(
+                CassandraOptions.CASSANDRA_RECONNECT_BASE_DELAY.name(), 10_000L);
+        conf.setProperty(
+                CassandraOptions.CASSANDRA_RECONNECT_MAX_DELAY.name(), 2_000L);
+        conf.setProperty(
+                CassandraOptions.CASSANDRA_RECONNECT_MAX_RETRIES.name(), 3);
+        conf.setProperty(
+                CassandraOptions.CASSANDRA_RECONNECT_INTERVAL.name(), 1_000L);
+        HugeConfig config = new HugeConfig(conf);
+        Assert.assertThrows(IllegalArgumentException.class, () ->
+                new CassandraSessionPool(config, "ks", "store"));
     }
 }