This is an automated email from the ASF dual-hosted git repository.
lukasz-antoniak pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/trunk by this push:
new be5080e3 CASSANALYTICS-179: Fix total timeout calculation for
getAllNodeSettings
be5080e3 is described below
commit be5080e3f907a65b8acc5ce7e9c4203978d88fde
Author: Lukasz Antoniak <[email protected]>
AuthorDate: Thu Jul 9 09:33:18 2026 +0200
CASSANALYTICS-179: Fix total timeout calculation for getAllNodeSettings
---
CHANGES.txt | 1 +
.../spark/bulkwriter/CassandraClusterInfo.java | 11 ++-
.../spark/bulkwriter/CassandraClusterInfoTest.java | 80 ++++++++++++++++++++++
3 files changed, 86 insertions(+), 6 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 7834ba38..1d645a06 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Fix total timeout calculation for getAllNodeSettings (CASSANALYTICS-179)
* Support vector data type (CASSANALYTICS-26)
* CDC batch-write mixing a CDC-enabled and CDC-disabled table drops the CDC
table's mutation (CASSANALYTICS-182)
* CdcState.ReplicaCountSerializer map-size overflow corrupts persisted CDC
state (CASSANALYTICS-184)
diff --git
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
index 1b1a4017..3f7523bc 100644
---
a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
+++
b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
@@ -502,12 +502,11 @@ public class CassandraClusterInfo implements ClusterInfo,
Closeable
+ "Cassandra version is
pre-computed on driver and broadcast to executors.");
}
- // Worst-case, the http client is configured for 1 worker pool.
- // In that case, each future can take the full retry delay * number of
retries,
- // and each instance will be processed serially.
- final long totalTimeout = conf.getSidecarRequestMaxRetryDelayMillis() *
- conf.getSidecarRequestRetries() *
- allNodeSettingFutures.size();
+ // Each of the retry attempts can take up to the request timeout plus
the max delay
+ // before the next retry. Requests to all instances run in parallel,
so the cluster-wide
+ // wait is bounded by a single node's worst case.
+ final long totalTimeout =
(TimeUnit.SECONDS.toMillis(conf.getSidecarRequestTimeoutSeconds()) +
conf.getSidecarRequestMaxRetryDelayMillis())
+ * conf.getSidecarRequestRetries();
List<NodeSettings> allNodeSettings =
FutureUtils.bestEffortGet(allNodeSettingFutures,
totalTimeout,
TimeUnit.MILLISECONDS);
diff --git
a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
index 5ed4c522..99b06712 100644
---
a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
+++
b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
@@ -21,12 +21,22 @@ package org.apache.cassandra.spark.bulkwriter;
import java.time.Duration;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import o.a.c.sidecar.client.shaded.common.response.NodeSettings;
import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
@@ -70,11 +80,55 @@ public class CassandraClusterInfoTest
"clusterId=null");
}
+ static Stream<Arguments> sidecarResponseDelays()
+ {
+ return Stream.of(
+ Arguments.of((Object) new int[] {100, 200, 300}), // all responses
within deadline
+ Arguments.of((Object) new int[] {500, 3000}) // single timeout
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("sidecarResponseDelays")
+ @Timeout(value = 2300, unit = TimeUnit.MILLISECONDS) // set timeout
slightly higher than deadline of (1000 + 100) * 2
+ void testSuccessfulGetAllNodeSettings(int[] responseDelayMillis)
+ {
+ BulkSparkConf conf = mockBulkSparkWithSidecarConf(1, 100, 2);
+ try (CassandraClusterInfo ci = new
MockClusterInfoForNodeSettings(conf, responseDelayMillis))
+ {
+ assertThatNoException()
+ .describedAs("Accept when at least one node responds within total
timeout")
+ .isThrownBy(ci::getAllNodeSettings);
+ }
+ }
+
+ @Test
+ void testTimeoutGetAllNodeSettings()
+ {
+ BulkSparkConf conf = mockBulkSparkWithSidecarConf(1, 100, 2);
+ try (CassandraClusterInfo ci = new
MockClusterInfoForNodeSettings(conf, 3000, 3300))
+ {
+ assertThatThrownBy(ci::getAllNodeSettings)
+ .describedAs("Raise error when no responses received within
timeout")
+ .isExactlyInstanceOf(RuntimeException.class)
+ .hasMessage("Unable to determine the node settings. 0/2 instances
available.");
+ }
+ }
+
public static CassandraClusterInfo mockClusterInfoForTimeSkewTest(int
allowanceMinutes, Instant remoteNow)
{
return new MockClusterInfoForTimeSkew(allowanceMinutes, remoteNow);
}
+ private BulkSparkConf mockBulkSparkWithSidecarConf(int
requestTimeoutSeconds, long maxRetryDelayMillis, int retryCount)
+ {
+ BulkSparkConf conf = mock(BulkSparkConf.class);
+
when(conf.getSidecarRequestTimeoutSeconds()).thenReturn(requestTimeoutSeconds);
+
when(conf.getSidecarRequestMaxRetryDelayMillis()).thenReturn(maxRetryDelayMillis);
+ when(conf.getSidecarRequestRetries()).thenReturn(retryCount);
+ return conf;
+ }
+
private static class MockClusterInfoForTimeSkew extends
CassandraClusterInfo
{
private CassandraContext cassandraContext;
@@ -107,4 +161,30 @@ public class CassandraClusterInfoTest
when(cassandraContext.sidecarPort()).thenReturn(9043);
}
}
+
+ private static class MockClusterInfoForNodeSettings extends
CassandraClusterInfo
+ {
+ MockClusterInfoForNodeSettings(BulkSparkConf conf, int...
responseDelayMillis)
+ {
+ super(conf);
+
+ allNodeSettingFutures.clear();
+ List<CompletableFuture<NodeSettings>> futures = new
ArrayList<>(responseDelayMillis.length);
+ for (int delay : responseDelayMillis)
+ {
+ CompletableFuture<NodeSettings> future =
CompletableFuture.supplyAsync(() -> {
+ Uninterruptibles.sleepUninterruptibly(delay,
TimeUnit.MILLISECONDS);
+ return mock(NodeSettings.class);
+ });
+ futures.add(future);
+ }
+ allNodeSettingFutures.addAll(futures);
+ }
+
+ @Override
+ protected CassandraContext buildCassandraContext()
+ {
+ return mock(CassandraContext.class);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
