Fix total timeout calculation for getAllNodeSettings

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)

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java

@@ -502,12 +502,11 @@ protected List<NodeSettings> getAllNodeSettings()
                                             + "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);

cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java

@@ -21,12 +21,22 @@
 
 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 @@ void testTimeSkewTooLarge()
                     "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 @@ private void mockCassandraContext(int allowanceMinutes, Instant remoteNow)
             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);
+        }
+    }
 }