Copilot commented on code in PR #2997:
URL: https://github.com/apache/hugegraph/pull/2997#discussion_r3142091084
##########
hugegraph-server/hugegraph-cassandra/src/main/java/org/apache/hugegraph/backend/store/cassandra/CassandraSessionPool.java:
##########
@@ -255,6 +391,56 @@ public boolean hasChanges() {
return this.batch.size() > 0;
}
+ /**
+ * 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. 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() {
+ if (!this.opened) {
+ return;
+ }
+ try {
+ if (this.session == null || this.session.isClosed()) {
+ this.session = null;
+ this.tryOpen();
+ }
+ if (this.session != null) {
+ this.session.execute(new
SimpleStatement(HEALTH_CHECK_CQL));
+ }
+ } catch (DriverException e) {
+ LOG.debug("Cassandra health-check failed, resetting session:
{}",
+ e.getMessage());
+ this.session = null;
Review Comment:
In `reconnectIfNeeded()`, the catch block sets `this.session = null` but
never closes the existing driver session, which can leak resources and
contradicts the javadoc (“discarded via reset()”). Call `reset()` (or at least
`close()` in a guarded way) when the health-check fails so the underlying
driver session is actually released.
```suggestion
this.reset();
```
##########
hugegraph-server/hugegraph-cassandra/src/main/java/org/apache/hugegraph/backend/store/cassandra/CassandraSessionPool.java:
##########
@@ -174,6 +228,11 @@ public void commitAsync() {
int processors = Math.min(statements.size(), 1023);
List<ResultSetFuture> results = new ArrayList<>(processors + 1);
for (Statement s : statements) {
+ // 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.
Review Comment:
The TODO comment in `commitAsync()` says callers must handle
“CompletableFuture failures”, but this method is using the Datastax
`ResultSetFuture` API (`getUninterruptibly()`), not `CompletableFuture`. Please
adjust the comment to match the actual async type/behavior to avoid confusion
for maintainers.
```suggestion
// must handle ResultSetFuture failures surfaced by
// getUninterruptibly(). A follow-up will wrap each future
// with retry semantics.
```
##########
hugegraph-server/hugegraph-cassandra/src/main/java/org/apache/hugegraph/backend/store/cassandra/CassandraSessionPool.java:
##########
@@ -197,15 +256,92 @@ public ResultSet query(Statement statement) {
}
public ResultSet execute(Statement statement) {
- return this.session.execute(statement);
+ return this.executeWithRetry(statement);
}
public ResultSet execute(String statement) {
- return this.session.execute(statement);
+ return this.executeWithRetry(new SimpleStatement(statement));
}
public ResultSet execute(String statement, Object... args) {
- return this.session.execute(statement, args);
+ return this.executeWithRetry(new SimpleStatement(statement, args));
+ }
+
+ /**
+ * Execute a statement, retrying on transient connectivity failures
+ * (NoHostAvailableException / OperationTimedOutException). The driver
+ * itself keeps retrying connections in the background via the
+ * 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 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) {
+ int retries = CassandraSessionPool.this.maxRetries;
+ long interval = CassandraSessionPool.this.retryInterval;
+ long maxDelay = CassandraSessionPool.this.retryMaxDelay;
+ 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) {
Review Comment:
`executeWithRetry()` retries on `OperationTimedOutException`. A timeout can
occur after the coordinator has already applied a mutation (or partially
applied a batch), so retrying can duplicate non-idempotent writes (e.g.,
counter increments in `CassandraTables.Counters.increaseCounter()`), producing
incorrect data. Consider limiting retries to connection-level failures like
`NoHostAvailableException`, or only retrying statements explicitly marked
idempotent (e.g., check `statement.isIdempotent()` / set idempotence on
statements) before retrying timeouts.
##########
hugegraph-server/hugegraph-cassandra/src/main/java/org/apache/hugegraph/backend/store/cassandra/CassandraSessionPool.java:
##########
@@ -174,6 +228,11 @@ public void commitAsync() {
int processors = Math.min(statements.size(), 1023);
List<ResultSetFuture> results = new ArrayList<>(processors + 1);
for (Statement s : statements) {
+ // 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);
Review Comment:
`commitAsync()` calls `this.session.executeAsync(...)` directly without
ensuring the driver session is open. Since `reset()` / `reconnectIfNeeded()`
can set `this.session` to null, this can now throw NPE at runtime. Consider
reopening lazily (similar to `executeWithRetry()`), or failing with a
`BackendException` with a clear message when the session is unavailable.
##########
hugegraph-server/hugegraph-cassandra/src/main/java/org/apache/hugegraph/backend/store/cassandra/CassandraOptions.java:
##########
@@ -130,4 +130,46 @@ public static synchronized CassandraOptions instance() {
positiveInt(),
12 * 60 * 60
);
+
+ public static final ConfigOption<Long> CASSANDRA_RECONNECT_BASE_DELAY =
+ new ConfigOption<>(
+ "cassandra.reconnect_base_delay",
+ "The base delay in milliseconds used by the driver's " +
+ "exponential reconnection policy when a Cassandra host " +
+ "becomes unreachable.",
+ rangeInt(100L, Long.MAX_VALUE),
+ 1000L
+ );
+
+ public static final ConfigOption<Long> CASSANDRA_RECONNECT_MAX_DELAY =
+ new ConfigOption<>(
+ "cassandra.reconnect_max_delay",
+ "The maximum delay in milliseconds used by the driver's " +
+ "exponential reconnection policy when a Cassandra host " +
+ "becomes unreachable.",
+ rangeInt(1000L, Long.MAX_VALUE),
+ 10_000L
+ );
+
Review Comment:
The defaults in code (max_delay=10_000ms, max_retries=3, interval=1000ms)
don’t match the PR description’s defaults table (max_delay=60000ms,
max_retries=10, interval=5000ms). Please align either the code/tests or update
the PR description so operators aren’t misled about the runtime behavior and
backoff characteristics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
