diff --git a/CHANGES.txt b/CHANGES.txt index 7834ba38b..1d645a060 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 1b1a40179..3f7523bc0 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 @@ protected List 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 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 5ed4c5221..99b06712f 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 @@ 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 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> futures = new ArrayList<>(responseDelayMillis.length); + for (int delay : responseDelayMillis) + { + CompletableFuture 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); + } + } }