You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by yc...@apache.org on 2025/02/16 16:11:33 UTC
(cassandra-sidecar) branch trunk updated: CASSSIDECAR-185: Adapt to cluster topology change for restore jobs (#171)
This is an automated email from the ASF dual-hosted git repository.
ycai pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
The following commit(s) were added to refs/heads/trunk by this push:
new 96b3c43e CASSSIDECAR-185: Adapt to cluster topology change for restore jobs (#171)
96b3c43e is described below
commit 96b3c43e32529d41a62649ce34acd56f883292f3
Author: Yifan Cai <yc...@apache.org>
AuthorDate: Sun Feb 16 08:11:27 2025 -0800
CASSSIDECAR-185: Adapt to cluster topology change for restore jobs (#171)
Patch by Yifan Cai; Reviewed by Francisco Guerrero for CASSSIDECAR-185
---
.circleci/config.yml | 12 +-
CHANGES.txt | 1 +
checkstyle.xml | 4 +-
.../sidecar/common/data/RestoreJobStatus.java | 27 ++-
.../request/data/CreateSliceRequestPayload.java | 20 +-
.../response/TokenRangeReplicasResponse.java | 6 +-
.../request/CreateSliceRequestPayloadTest.java | 4 +-
.../sidecar/common/server/StorageOperations.java | 5 +-
.../common/server/cluster/locator/TokenRange.java | 116 +++++++--
.../common/server/data/RestoreRangeStatus.java | 17 +-
.../sidecar/common/server/utils/StringUtils.java | 15 ++
.../exceptions/SchemaUnavailableException.java | 12 +-
.../cluster/locator/TokenRangeReplicasTest.java | 50 ++--
.../server/cluster/locator/TokenRangeTest.java | 95 ++++++--
.../common/server/data/RestoreRangeStatusTest.java | 45 ++--
.../sidecar/cluster/CassandraAdapterDelegate.java | 2 +-
.../cluster/locator/CachedLocalTokenRanges.java | 7 +-
.../cluster/locator/LocalTokenRangesProvider.java | 14 +-
.../sidecar/config/PeriodicTaskConfiguration.java | 3 +
.../config/yaml/PeriodicTaskConfigurationImpl.java | 5 +-
.../apache/cassandra/sidecar/db/RestoreRange.java | 67 +++--
.../sidecar/db/RestoreRangeDatabaseAccessor.java | 21 +-
.../apache/cassandra/sidecar/db/RestoreSlice.java | 100 +++-----
.../sidecar/db/RestoreSliceDatabaseAccessor.java | 7 +-
.../sidecar/db/schema/RestoreRangesSchema.java | 30 +--
.../sidecar/db/schema/RestoreSlicesSchema.java | 6 +-
.../sidecar/db/schema/SystemAuthSchema.java | 5 +-
...cker.java => RestoreJobConsistencyChecker.java} | 138 ++++++++---
.../sidecar/restore/RestoreJobDiscoverer.java | 270 +++++++++++++++++----
.../sidecar/restore/RestoreJobManager.java | 17 +-
.../sidecar/restore/RestoreJobManagerGroup.java | 30 +++
.../sidecar/restore/RestoreJobProgress.java | 18 ++
.../sidecar/restore/RestoreJobProgressTracker.java | 50 +++-
.../sidecar/restore/RestoreProcessor.java | 16 +-
.../sidecar/restore/RestoreRangeTask.java | 2 +-
.../restore/RingTopologyChangeListener.java | 39 +++
.../sidecar/restore/RingTopologyRefresher.java | 267 +++++++++++++++++---
.../routes/restore/RestoreJobProgressHandler.java | 6 +-
.../db/RestoreRangeDatabaseAccessorIntTest.java | 8 +-
.../sidecar/restore/RestoreJobTestUtils.java | 173 +++++++++++++
.../RestoreJobDiscovererNodeJoinedIntTest.java | 164 +++++++++++++
.../RestoreJobDiscovererNodeJoiningIntTest.java | 190 +++++++++++++++
.../RestoreJobDiscovererNodeLeavingIntTest.java | 178 ++++++++++++++
.../RestoreJobDiscovererNodeLeftIntTest.java | 155 ++++++++++++
.../RestoreJobDiscovererNodeMovedIntTest.java | 177 ++++++++++++++
.../RestoreJobDiscovererNodeMovingIntTest.java | 182 ++++++++++++++
.../tokenrange/BaseTokenRangeIntegrationTest.java | 1 +
.../sidecar/routes/tokenrange/JoiningBaseTest.java | 1 +
.../tokenrange/JoiningTestDoubleCluster.java | 2 +-
.../routes/tokenrange/JoiningTestMultiDC.java | 2 +-
.../JoiningTestMultiDCSingleReplicated.java | 2 +-
.../tokenrange/JoiningTestMultipleNodes.java | 2 +-
.../routes/tokenrange/JoiningTestSingleNode.java | 53 +---
.../sidecar/routes/tokenrange/LeavingTest.java | 51 +---
.../sidecar/routes/tokenrange/MovingBaseTest.java | 1 +
.../sidecar/routes/tokenrange/MovingTest.java | 61 +----
.../routes/tokenrange/ReplacementMultiDCTest.java | 2 +-
.../sidecar/routes/tokenrange/ReplacementTest.java | 3 +-
.../sidecar/testing/BootstrapBBUtils.java | 2 +-
.../tokenrange => testing}/TestTokenSupplier.java | 19 +-
.../testing/bytebuddy/BBHelperJoiningNode.java | 64 +++++
.../testing/bytebuddy/BBHelperLeavingNode.java | 75 ++++++
.../testing/bytebuddy/BBHelperMovingNode.java | 77 ++++++
.../cassandra/sidecar/db/RestoreSliceTest.java | 72 +++---
.../cassandra/sidecar/db/SidecarSchemaTest.java | 6 +-
....java => RestoreJobConsistencyCheckerTest.java} | 85 ++++++-
.../sidecar/restore/RestoreJobDiscovererTest.java | 5 +-
.../sidecar/restore/RestoreJobManagerTest.java | 68 +++++-
.../sidecar/restore/RestoreProcessorTest.java | 74 ++++--
.../sidecar/restore/RestoreRangeTest.java | 18 +-
.../sidecar/restore/RingTopologyRefresherTest.java | 47 +++-
.../routes/restore/BaseRestoreJobTests.java | 10 +-
.../config/yaml/TestServiceConfiguration.java | 5 +-
spotbugs-exclude.xml | 14 ++
vertx-auth-mtls/build.gradle | 3 +
75 files changed, 2961 insertions(+), 640 deletions(-)
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 1421a345..48bb955e 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -142,8 +142,8 @@ jobs:
docker:
- image: cimg/openjdk:11.0
environment:
- INTEGRATION_MAX_PARALLEL_FORKS: 1
- INTEGRATION_MAX_HEAP_SIZE: "7000M"
+ INTEGRATION_MAX_PARALLEL_FORKS: 2
+ INTEGRATION_MAX_HEAP_SIZE: "3600M"
resource_class: large
steps:
- install_common
@@ -194,8 +194,8 @@ jobs:
docker:
- image: cimg/openjdk:11.0
environment:
- INTEGRATION_MAX_PARALLEL_FORKS: 1
- INTEGRATION_MAX_HEAP_SIZE: "7000M"
+ INTEGRATION_MAX_PARALLEL_FORKS: 2
+ INTEGRATION_MAX_HEAP_SIZE: "3600M"
resource_class: large
steps:
- install_common
@@ -246,8 +246,8 @@ jobs:
docker:
- image: cimg/openjdk:11.0
environment:
- INTEGRATION_MAX_PARALLEL_FORKS: 1
- INTEGRATION_MAX_HEAP_SIZE: "7000M"
+ INTEGRATION_MAX_PARALLEL_FORKS: 2
+ INTEGRATION_MAX_HEAP_SIZE: "3600M"
resource_class: large
steps:
- install_common
diff --git a/CHANGES.txt b/CHANGES.txt
index 4f20484a..f38a692f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
1.0.0
-----
+ * Adapt to cluster topology change for restore jobs (CASSSIDECAR-185)
* Fix PeriodicTaskExecutor double execution due to race from reschedule (CASSSIDECAR-210)
* Upgrade Netty to 4.1.118.Final and Vert.x to 4.5.13 Version (CASSSIDECAR-207)
* Fix missing field for INDEX_STATUS in GossipInfo (CASSSIDECAR-195)
diff --git a/checkstyle.xml b/checkstyle.xml
index 0b19a07e..7dd42537 100644
--- a/checkstyle.xml
+++ b/checkstyle.xml
@@ -287,7 +287,9 @@ page at http://checkstyle.sourceforge.net/config.html -->
some other variants which we don't publicized to promote consistency).
-->
<property name="reliefPattern"
- value="fall through|Fall through|fallthru|Fallthru|falls through|Falls through|fallthrough|Fallthrough|No break|NO break|no break|continue on" />
+ value="fall through|Fall through|fallthru|Fallthru|falls through
+ |Falls through|fallthrough|Fallthrough|No break|NO break|no break
+ |continue on|no break by design" />
<property name="severity" value="error" />
</module>
diff --git a/client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobStatus.java b/client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobStatus.java
index 2f3ce544..9074ccb5 100644
--- a/client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobStatus.java
+++ b/client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobStatus.java
@@ -23,13 +23,38 @@ package org.apache.cassandra.sidecar.common.data;
*/
public enum RestoreJobStatus
{
+ /**
+ * The external controller creates the RestoreJob
+ */
CREATED,
+ /**
+ * The external controller updates the status of the RestoreJob to STAGE_READY.
+ * It indicates that all relevant slices of the RestoreJob have been uploaded and ready to be consumed.
+ */
STAGE_READY,
+ /**
+ * All relevant slices are staged and the staged data satisfies the consistency requirement of the RestoreJob.
+ * The external controller updates the status of the RestoreJob to STAGED.
+ */
STAGED,
+ /**
+ * The external controller updates the status of the RestoreJob to IMPORT_READY
+ * It indicates that all staged data now are ready to be imported into Cassandra.
+ */
IMPORT_READY,
- @Deprecated // replaced by ABORTED
+ /**
+ * @deprecated replaced by {@link #ABORTED}
+ */
FAILED,
+ /**
+ * The external controller aborts the RestoreJob due to some failure, e.g. consistency not satisfied, timeout, etc.
+ * The RestoreJob can also be aborted by Sidecar, if and only if, the external controller fails to renew the lease for certain long duration.
+ * Such RestoreJobs are aborted due to expiry.
+ */
ABORTED,
+ /**
+ * The external controller marks the RestoreJob as succeeded.
+ */
SUCCEEDED;
public boolean isFinal()
diff --git a/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateSliceRequestPayload.java b/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateSliceRequestPayload.java
index 1afb088e..d0e72c5a 100644
--- a/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateSliceRequestPayload.java
+++ b/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateSliceRequestPayload.java
@@ -50,8 +50,8 @@ public class CreateSliceRequestPayload
private final String bucket;
private final String key;
private final String checksum;
- private final BigInteger startToken;
- private final BigInteger endToken;
+ private final BigInteger firstToken; // inclusive
+ private final BigInteger endToken; // inclusive
// -- Optional fields - using Objects
private final Long uncompressedSize;
private final Long compressedSize;
@@ -63,7 +63,7 @@ public class CreateSliceRequestPayload
@JsonProperty(SLICE_STORAGE_BUCKET) String bucket,
@JsonProperty(SLICE_STORAGE_KEY) String key,
@JsonProperty(SLICE_CHECKSUM) String checksum,
- @JsonProperty(SLICE_START_TOKEN) BigInteger startToken,
+ @JsonProperty(SLICE_START_TOKEN) BigInteger firstToken,
@JsonProperty(SLICE_END_TOKEN) BigInteger endToken,
@JsonProperty(SLICE_UNCOMPRESSED_SIZE) Long uncompressedSize,
@JsonProperty(SLICE_COMPRESSED_SIZE) Long compressedSize)
@@ -72,7 +72,7 @@ public class CreateSliceRequestPayload
&& checksum != null
&& bucket != null
&& key != null
- && startToken != null
+ && firstToken != null
&& endToken != null,
"Invalid create slice request payload");
Preconditions.checkArgument(bucketId < Short.MAX_VALUE && bucketId >= 0,
@@ -83,7 +83,7 @@ public class CreateSliceRequestPayload
this.bucket = bucket;
this.key = key;
this.checksum = checksum;
- this.startToken = startToken;
+ this.firstToken = firstToken;
this.endToken = endToken;
this.uncompressedSize = uncompressedSize;
this.compressedSize = compressedSize;
@@ -131,16 +131,16 @@ public class CreateSliceRequestPayload
}
/**
- * @return start token of slice
+ * @return first token of slice; inclusive
*/
@JsonProperty(SLICE_START_TOKEN)
- public BigInteger startToken()
+ public BigInteger firstToken()
{
- return startToken;
+ return firstToken;
}
/**
- * @return end token of slice
+ * @return end token of slice; inclusive
*/
@JsonProperty(SLICE_END_TOKEN)
public BigInteger endToken()
@@ -201,7 +201,7 @@ public class CreateSliceRequestPayload
SLICE_STORAGE_BUCKET + "='" + bucket + "', " +
SLICE_STORAGE_KEY + "='" + key + "', " +
SLICE_CHECKSUM + "='" + checksum + "', " +
- SLICE_START_TOKEN + "='" + startToken + "', " +
+ SLICE_START_TOKEN + "='" + firstToken + "', " +
SLICE_END_TOKEN + "='" + endToken + "', " +
SLICE_COMPRESSED_SIZE + "='" + compressedSize + "', " +
SLICE_UNCOMPRESSED_SIZE + "='" + uncompressedSize + "'}";
diff --git a/client-common/src/main/java/org/apache/cassandra/sidecar/common/response/TokenRangeReplicasResponse.java b/client-common/src/main/java/org/apache/cassandra/sidecar/common/response/TokenRangeReplicasResponse.java
index 742fc4dd..e37ba1cd 100644
--- a/client-common/src/main/java/org/apache/cassandra/sidecar/common/response/TokenRangeReplicasResponse.java
+++ b/client-common/src/main/java/org/apache/cassandra/sidecar/common/response/TokenRangeReplicasResponse.java
@@ -83,7 +83,9 @@ public class TokenRangeReplicasResponse
*/
public static class ReplicaInfo
{
+ // exclusive
private final String start;
+ // inclusive
private final String end;
private final Map<String, List<String>> replicasByDatacenter;
@@ -97,7 +99,7 @@ public class TokenRangeReplicasResponse
}
/**
- * @return the start value of the token range
+ * @return the start value of the token range, exclusive
*/
@JsonProperty("start")
public String start()
@@ -106,7 +108,7 @@ public class TokenRangeReplicasResponse
}
/**
- * @return the end value of the token range
+ * @return the end value of the token range, inclusive
*/
@JsonProperty("end")
public String end()
diff --git a/client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateSliceRequestPayloadTest.java b/client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateSliceRequestPayloadTest.java
index 8d07f4cf..d9064e96 100644
--- a/client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateSliceRequestPayloadTest.java
+++ b/client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateSliceRequestPayloadTest.java
@@ -52,7 +52,7 @@ class CreateSliceRequestPayloadTest
assertThat(req.checksum()).isEqualTo("12321abc");
assertThat(req.bucket()).isEqualTo("myBucket");
assertThat(req.key()).isEqualTo("/path/to/object");
- assertThat(req.startToken()).isEqualTo(BigInteger.ONE);
+ assertThat(req.firstToken()).isEqualTo(BigInteger.ONE);
assertThat(req.endToken()).isEqualTo(BigInteger.TEN);
assertThat(req.compressedSize())
.describedAs("size is not present in the json")
@@ -85,7 +85,7 @@ class CreateSliceRequestPayloadTest
assertThat(test.bucket()).isEqualTo("bucket");
assertThat(test.key()).isEqualTo("key");
assertThat(test.checksum()).isEqualTo("checksum");
- assertThat(test.startToken()).isEqualTo(BigInteger.ONE);
+ assertThat(test.firstToken()).isEqualTo(BigInteger.ONE);
assertThat(test.endToken()).isEqualTo(BigInteger.TEN);
assertThat(test.uncompressedSize()).isEqualTo(234L);
assertThat(test.compressedSize()).isEqualTo(123L);
diff --git a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/StorageOperations.java b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/StorageOperations.java
index 0a5c16cd..58ed83ec 100644
--- a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/StorageOperations.java
+++ b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/StorageOperations.java
@@ -32,7 +32,6 @@ import org.jetbrains.annotations.Nullable;
/**
* An interface that defines interactions with the storage system in Cassandra.
- * TODO: relocate to server-common
*/
public interface StorageOperations
{
@@ -62,6 +61,8 @@ public interface StorageOperations
* @param keyspace keyspace to check the data ownership; Cassandra selects a keyspace if null value is passed.
* @return ring view
* @throws UnknownHostException when hostname of peer Cassandra nodes cannot be resolved
+ *
+ * TODO: refactor. Do not return http response payload object from this layer.
*/
RingResponse ring(@Nullable Name keyspace) throws UnknownHostException;
@@ -71,6 +72,8 @@ public interface StorageOperations
* @param keyspace the keyspace in the Cassandra database
* @param partitioner token partitioner used for token assignment
* @return token range to read and write replica mappings
+ *
+ * TODO: refactor. Do not return http response payload object from this layer.
*/
TokenRangeReplicasResponse tokenRangeReplicas(@NotNull Name keyspace,
@NotNull String partitioner);
diff --git a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRange.java b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRange.java
index 54df89ed..8f1e61c2 100644
--- a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRange.java
+++ b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRange.java
@@ -21,20 +21,21 @@ package org.apache.cassandra.sidecar.common.server.cluster.locator;
import java.math.BigInteger;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
import java.util.stream.Collectors;
import com.google.common.collect.Range;
+import com.google.common.collect.RangeSet;
+import com.google.common.collect.TreeRangeSet;
import com.datastax.driver.core.DataType;
-import org.jetbrains.annotations.Nullable;
/**
* Range: (start, end] - start exclusive and end inclusive
*/
public class TokenRange
{
- private final Range<Token> range;
- private volatile Token firstToken = null;
+ public final Range<Token> range;
/**
* Unwrap the java driver's token range if necessary and convert the unwrapped ranges list.
@@ -85,6 +86,29 @@ public class TokenRange
}
}
+ /**
+ * Diff the two set of {@link TokenRange}s. The connected token ranges in each set are merged before diffing.
+ * The result {@link SymmetricDiffResult#onlyInLeft} contains the token ranges that are only in the input {@code left} token range set, and
+ * the result {@link SymmetricDiffResult#onlyInRight} contains the token ranges that are only in the input {@code right} token range set.
+ * @param left token range set
+ * @param right token range set
+ * @return {@link SymmetricDiffResult} where its left contains the token ranges that are only in the input {@code left} token range set, and
+ * its right contains the token ranges that are only in the input {@code right} token range set.
+ */
+ @SuppressWarnings("UnstableApiUsage")
+ public static SymmetricDiffResult symmetricDiff(Set<TokenRange> left, Set<TokenRange> right)
+ {
+ RangeSet<Token> mergedLeft = TreeRangeSet.create();
+ RangeSet<Token> mergedRight = TreeRangeSet.create();
+ left.forEach(r -> mergedLeft.add(r.range));
+ right.forEach(r -> mergedRight.add(r.range));
+ RangeSet<Token> resultLeft = TreeRangeSet.create(mergedLeft);
+ resultLeft.removeAll(mergedRight);
+ RangeSet<Token> resultRight = TreeRangeSet.create(mergedRight);
+ resultRight.removeAll(mergedLeft);
+ return new SymmetricDiffResult(resultLeft.asRanges(), resultRight.asRanges());
+ }
+
public TokenRange(long start, long end)
{
this(BigInteger.valueOf(start), BigInteger.valueOf(end));
@@ -108,6 +132,14 @@ public class TokenRange
return range.lowerEndpoint();
}
+ /**
+ * @return start token as {@link BigInteger}. It is not enclosed in the range.
+ */
+ public BigInteger startAsBigInt()
+ {
+ return range.lowerEndpoint().toBigInteger();
+ }
+
/**
* @return end token. It is the last token enclosed in the range.
*/
@@ -117,22 +149,11 @@ public class TokenRange
}
/**
- * @return the first token enclosed in the range. It returns null if the range is empty, e.g. (v, v]
+ * @return end token as {@link BigInteger}. It is the last token enclosed in the range.
*/
- @Nullable
- public Token firstToken()
+ public BigInteger endAsBigInt()
{
- if (range.isEmpty())
- {
- return null;
- }
-
- // it is ok to race
- if (firstToken == null)
- {
- firstToken = range.lowerEndpoint().increment();
- }
- return firstToken;
+ return range.upperEndpoint().toBigInteger();
}
/**
@@ -145,22 +166,37 @@ public class TokenRange
}
/**
- * Two ranges are overlapping when their intersection is non-empty. For example,
+ * Two ranges are intersecting when their intersection is non-empty. For example,
+ * <p>Ranges {@code (0, 3]} and {@code (1, 4]} are overlapping. The intersection is {@code (1, 3]}
+ * <p>Ranges {@code (0, 3]} and {@code (5, 7]} are not overlapping, as there is no intersection
+ * <p>Ranges {@code (0, 3]} and {@code (3, 5]} are not overlapping, as the intersection {@code (3, 3]} is empty
*
- * Ranges (0, 3] and (1, 4] are overlapping. The intersection is (1, 3]
- * Ranges (0, 3] and (5, 7] are not overlapping, as there is no intersection
- * Ranges (0, 3] and (3, 5] are not overlapping, as the intersection (3, 3] is empty
+ * <p>Note that the semantics is different from {@link Range#isConnected(Range)}
*
- * Note that the semantics is different from {@link Range#isConnected(Range)}
- *
- * @return true if this range overlaps with the other range; otherwise, false
+ * @return true if this range intersects with the other range; otherwise, false
*/
- public boolean overlaps(TokenRange other)
+ public boolean intersects(TokenRange other)
{
return this.range.lowerEndpoint().compareTo(other.range.upperEndpoint()) < 0
&& other.range.lowerEndpoint().compareTo(this.range.upperEndpoint()) < 0;
}
+ /**
+ * Two ranges connect with each other when 1) they overlap or 2) their ends are connected.
+ * <p>For 1), refer to {@link #intersects(TokenRange)}
+ * <p>For 2), see the following examples. The ranges {@code (0, 3]} and {@code (3, 5]} are connected.
+ * The ranges {@code (0, 3]} and {@code (4, 6]} are not connected.
+ *
+ * <p> Note that it is implemented using {@link Range#isConnected(Range)}
+ *
+ * @param other the other range to check
+ * @return true if this range connects with the other range; otherwise, false
+ */
+ public boolean connectsWith(TokenRange other)
+ {
+ return this.range.isConnected(other.range);
+ }
+
/**
* @return the range that is enclosed by both this and the other ranges.
*/
@@ -202,4 +238,34 @@ public class TokenRange
{
return range.hashCode();
}
+
+ @Override
+ public String toString()
+ {
+ return "TokenRange(" +
+ range.lowerEndpoint().toBigInteger() + ", " +
+ range.upperEndpoint().toBigInteger() + ']';
+ }
+
+ /**
+ * Pair of {@link TokenRange} sets that represent the result of symmetric diff
+ */
+ public static class SymmetricDiffResult
+ {
+ public final Set<TokenRange> onlyInLeft;
+ public final Set<TokenRange> onlyInRight;
+
+ private SymmetricDiffResult(Set<Range<Token>> onlyInLeft, Set<Range<Token>> onlyInRight)
+ {
+ this.onlyInLeft = toUnmodifiableSet(onlyInLeft);
+ this.onlyInRight = toUnmodifiableSet(onlyInRight);
+ }
+
+ private static Set<TokenRange> toUnmodifiableSet(Set<Range<Token>> ranges)
+ {
+ return ranges.stream()
+ .map(r -> new TokenRange(r.lowerEndpoint(), r.upperEndpoint()))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+ }
}
diff --git a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/data/RestoreRangeStatus.java b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/data/RestoreRangeStatus.java
index bc4ec956..b260fdbb 100644
--- a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/data/RestoreRangeStatus.java
+++ b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/data/RestoreRangeStatus.java
@@ -29,6 +29,9 @@ import org.apache.cassandra.sidecar.common.utils.Preconditions;
*/
public enum RestoreRangeStatus
{
+ /**
+ * Data of the restore range is imported into Cassandra
+ */
SUCCEEDED,
/**
* Failed is caused by any unrecoverable exception during processing the range
@@ -38,8 +41,18 @@ public enum RestoreRangeStatus
* Aborted is caused by controller command
*/
ABORTED,
- STAGED(SUCCEEDED, FAILED, ABORTED),
- CREATED(STAGED, FAILED, ABORTED);
+ /**
+ * Discarded is a special terminal status, which is caused by losing the token range due to Cassandra cluster topology change
+ */
+ DISCARDED,
+ /**
+ * Data of the restore range is staged on local disk
+ */
+ STAGED(SUCCEEDED, FAILED, ABORTED, DISCARDED),
+ /**
+ * Restore range is created to be processed
+ */
+ CREATED(STAGED, FAILED, ABORTED, DISCARDED);
// Do not use EnumSet, since validTargetStatuses is assigned on constructing and enums are not available yet.
private final Set<RestoreRangeStatus> validTargetStatusSet;
diff --git a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/utils/StringUtils.java b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/utils/StringUtils.java
index 8cb38dc6..055487d4 100644
--- a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/utils/StringUtils.java
+++ b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/utils/StringUtils.java
@@ -42,4 +42,19 @@ public class StringUtils
{
return HostAndPort.fromParts(address.getAddress().getHostAddress(), address.getPort()).toString();
}
+
+ /**
+ * Similar to {@link #cassandraFormattedHostAndPort(InetSocketAddress)}, but use ip and port to
+ * create the address string representing the Cassandra server.
+ * @param ip ip address
+ * @param port port
+ * @return the string representation of the address with the same signature as represented by
+ * the Cassandra server
+ */
+ @SuppressWarnings("UnstableApiUsage")
+ public static String cassandraFormattedHostAndPort(String ip, int port)
+ {
+ return HostAndPort.fromParts(ip, port).toString();
+
+ }
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/exceptions/SchemaUnavailableException.java b/server-common/src/main/java/org/apache/cassandra/sidecar/exceptions/SchemaUnavailableException.java
similarity index 73%
rename from server/src/main/java/org/apache/cassandra/sidecar/exceptions/SchemaUnavailableException.java
rename to server-common/src/main/java/org/apache/cassandra/sidecar/exceptions/SchemaUnavailableException.java
index 0ea9ed91..a8ad3baa 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/exceptions/SchemaUnavailableException.java
+++ b/server-common/src/main/java/org/apache/cassandra/sidecar/exceptions/SchemaUnavailableException.java
@@ -19,18 +19,18 @@
package org.apache.cassandra.sidecar.exceptions;
/**
- * Exception thrown when {@link org.apache.cassandra.sidecar.db.schema.TableSchema} is not prepared or expected
- * operations are unavailable.
+ * Exception thrown when {@link org.apache.cassandra.sidecar.db.schema.TableSchema} does not exist.
+ * For instance, the connected Cassandra no longer has such table
*/
public class SchemaUnavailableException extends RuntimeException
{
- public SchemaUnavailableException(String message)
+ public SchemaUnavailableException(String keyspace, String table)
{
- super(message);
+ super(makeErrorMessage(keyspace, table));
}
- public SchemaUnavailableException(String message, Throwable cause)
+ private static String makeErrorMessage(String keyspace, String table)
{
- super(message, cause);
+ return "Table " + keyspace + '.' + table + " does not exist";
}
}
diff --git a/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRangeReplicasTest.java b/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRangeReplicasTest.java
index 3634bb15..8615349b 100644
--- a/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRangeReplicasTest.java
+++ b/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRangeReplicasTest.java
@@ -46,7 +46,7 @@ public class TokenRangeReplicasTest
LOGGER.info("Input:" + simpleList);
List<TokenRangeReplicas> rangeList = TokenRangeReplicas.normalize(simpleList);
LOGGER.info("Result:" + rangeList);
- assertThat(hasOverlaps(rangeList)).isFalse();
+ assertThat(hasIntersects(rangeList)).isFalse();
assertThat(simpleList).containsExactlyInAnyOrderElementsOf(rangeList);
}
@@ -58,7 +58,7 @@ public class TokenRangeReplicasTest
List<TokenRangeReplicas> rangeList = TokenRangeReplicas.normalize(subRangeList);
LOGGER.info("Result:" + rangeList);
- assertThat(hasOverlaps(rangeList)).isFalse();
+ assertThat(hasIntersects(rangeList)).isFalse();
// Validate that we have 1 additional list as a result of the splits
assertThat(rangeList).hasSize(subRangeList.size() + 1);
@@ -89,14 +89,14 @@ public class TokenRangeReplicasTest
Iterator<TokenRangeReplicas> iter = subRangeList.iterator();
TokenRangeReplicas curr = iter.next();
TokenRangeReplicas next = iter.next();
- assertThat(hasOverlaps(subRangeList)).isTrue();
+ assertThat(hasIntersects(subRangeList)).isTrue();
LOGGER.info("Input:" + subRangeList);
TokenRangeReplicas.processIntersectingRanges(output,
iter,
curr,
next);
LOGGER.info("Result:" + output);
- assertThat(hasOverlaps(output)).isFalse();
+ assertThat(hasIntersects(output)).isFalse();
assertThat(output).hasSize(6);
List<TokenRangeReplicas> expectedExists =
TokenRangeReplicas.generateTokenRangeReplicas(Token.from("10"),
@@ -132,14 +132,14 @@ public class TokenRangeReplicasTest
Iterator<TokenRangeReplicas> iter = subRangeList.iterator();
TokenRangeReplicas curr = iter.next();
TokenRangeReplicas next = iter.next();
- assertThat(hasOverlaps(subRangeList)).isTrue();
+ assertThat(hasIntersects(subRangeList)).isTrue();
LOGGER.info("Input:" + subRangeList);
TokenRangeReplicas.processIntersectingRanges(output,
iter,
curr,
next);
LOGGER.info("Result:" + output);
- assertThat(hasOverlaps(output)).isFalse();
+ assertThat(hasIntersects(output)).isFalse();
assertThat(output).hasSize(5);
List<TokenRangeReplicas> expectedExists =
TokenRangeReplicas.generateTokenRangeReplicas(Token.from("10"),
@@ -175,14 +175,14 @@ public class TokenRangeReplicasTest
Iterator<TokenRangeReplicas> iter = subRangeList.iterator();
TokenRangeReplicas curr = iter.next();
TokenRangeReplicas next = iter.next();
- assertThat(hasOverlaps(subRangeList)).isTrue();
+ assertThat(hasIntersects(subRangeList)).isTrue();
LOGGER.info("Input:" + subRangeList);
TokenRangeReplicas.processIntersectingRanges(output,
iter,
curr,
next);
LOGGER.info("Result:" + output);
- assertThat(hasOverlaps(output)).isFalse();
+ assertThat(hasIntersects(output)).isFalse();
assertThat(output).hasSize(6);
List<TokenRangeReplicas> expectedExists =
TokenRangeReplicas.generateTokenRangeReplicas(Token.from("10"),
@@ -218,7 +218,7 @@ public class TokenRangeReplicasTest
LOGGER.info("Input:" + partialOverlapList);
List<TokenRangeReplicas> rangeList = TokenRangeReplicas.normalize(partialOverlapList);
LOGGER.info("Result:" + rangeList);
- assertThat(hasOverlaps(rangeList)).isFalse();
+ assertThat(hasIntersects(rangeList)).isFalse();
// Validate that we have 1 additional list as a result of the splits
assertThat(rangeList).hasSize(partialOverlapList.size() + 1);
@@ -264,7 +264,7 @@ public class TokenRangeReplicasTest
LOGGER.info("Input:" + multiOverlapList);
List<TokenRangeReplicas> rangeList = TokenRangeReplicas.normalize(multiOverlapList);
LOGGER.info("Result:" + rangeList);
- assertThat(hasOverlaps(rangeList)).isFalse();
+ assertThat(hasIntersects(rangeList)).isFalse();
// Validate that we have 1 additional list as a result of the splits
assertThat(rangeList).hasSize(multiOverlapList.size() + 1);
@@ -299,7 +299,7 @@ public class TokenRangeReplicasTest
LOGGER.info("Input:" + overlapList);
List<TokenRangeReplicas> rangeList = TokenRangeReplicas.normalize(overlapList);
LOGGER.info("Result:" + rangeList);
- assertThat(hasOverlaps(rangeList)).isFalse();
+ assertThat(hasIntersects(rangeList)).isFalse();
// Validate that we have 1 additional ranges as a result of the merges and splits
assertThat(rangeList).hasSize(overlapList.size() + 1);
@@ -339,7 +339,7 @@ public class TokenRangeReplicasTest
LOGGER.info("Input:" + wrappedOverlapList);
List<TokenRangeReplicas> rangeList = TokenRangeReplicas.normalize(wrappedOverlapList);
LOGGER.info("Result:" + rangeList);
- assertThat(hasOverlaps(rangeList)).isFalse();
+ assertThat(hasIntersects(rangeList)).isFalse();
// Split & Merge should result in same number of ranges
// (-1, 10] and (-1, 20] results in (-1, 10] and (10, 20]
@@ -382,7 +382,7 @@ public class TokenRangeReplicasTest
LOGGER.info("Input:" + wrappedOverlapList);
List<TokenRangeReplicas> rangeList = TokenRangeReplicas.normalize(wrappedOverlapList);
LOGGER.info("Result:" + rangeList);
- assertThat(hasOverlaps(rangeList)).isFalse();
+ assertThat(hasIntersects(rangeList)).isFalse();
// Split & Merge should result in same number of ranges
// (-1, 5] and (-1, 10] results in (-1, 5] and (5, 10]
@@ -466,7 +466,7 @@ public class TokenRangeReplicasTest
LOGGER.info("Input:" + createdList);
List<TokenRangeReplicas> rangeList = TokenRangeReplicas.normalize(createdList);
LOGGER.info("Result:" + rangeList);
- assertThat(hasOverlaps(rangeList)).isFalse();
+ assertThat(hasIntersects(rangeList)).isFalse();
assertThat(rangeList).hasSize(5);
List<TokenRangeReplicas> expectedExists = TokenRangeReplicas.generateTokenRangeReplicas(
@@ -547,7 +547,7 @@ public class TokenRangeReplicasTest
assertThat(intersectsWithRanges(range6.get(0), range5)).isTrue();
}
- private boolean hasOverlaps(List<TokenRangeReplicas> rangeList)
+ private boolean hasIntersects(List<TokenRangeReplicas> rangeList)
{
Collections.sort(rangeList);
for (int c = 0, i = 1; i < rangeList.size(); i++)
@@ -590,7 +590,7 @@ public class TokenRangeReplicasTest
Partitioners.RANDOM,
new HashSet<>(Arrays.asList("h6", "h7"))).get(0)
);
- assertThat(hasOverlaps(simpleList)).isFalse();
+ assertThat(hasIntersects(simpleList)).isFalse();
return simpleList;
}
@@ -615,7 +615,7 @@ public class TokenRangeReplicasTest
Partitioners.RANDOM,
new HashSet<>(Arrays.asList("h9"))).get(0)
);
- assertThat(hasOverlaps(rangeWithOverlaps)).isTrue();
+ assertThat(hasIntersects(rangeWithOverlaps)).isTrue();
return rangeWithOverlaps;
}
@@ -640,7 +640,7 @@ public class TokenRangeReplicasTest
Partitioners.RANDOM,
new HashSet<>(Arrays.asList("h9"))).get(0)
);
- assertThat(hasOverlaps(rangeWithOverlaps)).isTrue();
+ assertThat(hasIntersects(rangeWithOverlaps)).isTrue();
return rangeWithOverlaps;
}
@@ -665,7 +665,7 @@ public class TokenRangeReplicasTest
Partitioners.RANDOM,
new HashSet<>(Arrays.asList("h9"))).get(0)
);
- assertThat(hasOverlaps(rangeWithOverlaps)).isTrue();
+ assertThat(hasIntersects(rangeWithOverlaps)).isTrue();
return rangeWithOverlaps;
}
@@ -701,7 +701,7 @@ public class TokenRangeReplicasTest
createdList.addAll(wrappedList);
assertThat(hasWrappedRange(createdList)).isFalse();
- assertThat(hasOverlaps(createdList)).isTrue();
+ assertThat(hasIntersects(createdList)).isTrue();
return createdList;
}
@@ -730,7 +730,7 @@ public class TokenRangeReplicasTest
Partitioners.RANDOM,
new HashSet<>(Arrays.asList("h9"))).get(0)
);
- assertThat(hasOverlaps(rangeWithOverlaps)).isTrue();
+ assertThat(hasIntersects(rangeWithOverlaps)).isTrue();
return rangeWithOverlaps;
}
@@ -754,7 +754,7 @@ public class TokenRangeReplicasTest
Partitioners.RANDOM,
new HashSet<>(Arrays.asList("h9"))).get(0)
);
- assertThat(hasOverlaps(rangeWithOverlaps)).isTrue();
+ assertThat(hasIntersects(rangeWithOverlaps)).isTrue();
return rangeWithOverlaps;
}
@@ -779,7 +779,7 @@ public class TokenRangeReplicasTest
Partitioners.RANDOM,
new HashSet<>(Arrays.asList("h9"))).get(0)
);
- assertThat(hasOverlaps(rangeWithOverlaps)).isTrue();
+ assertThat(hasIntersects(rangeWithOverlaps)).isTrue();
return rangeWithOverlaps;
}
@@ -813,7 +813,7 @@ public class TokenRangeReplicasTest
assertThat(wrappedRange).hasSize(2);
// We should not have wrapped ranges as generateTokenRangeReplicas unwraps them, and we validate this below
assertThat(hasWrappedRange(createdList)).isFalse();
- assertThat(hasOverlaps(createdList)).isTrue();
+ assertThat(hasIntersects(createdList)).isTrue();
return createdList;
}
@@ -847,7 +847,7 @@ public class TokenRangeReplicasTest
assertThat(wrappedRange2).hasSize(2);
// We should not have wrapped ranges as generateTokenRangeReplicas unwraps them, and we validate this below
assertThat(hasWrappedRange(createdList)).isFalse();
- assertThat(hasOverlaps(createdList)).isTrue();
+ assertThat(hasIntersects(createdList)).isTrue();
return createdList;
}
diff --git a/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRangeTest.java b/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRangeTest.java
index 103ff934..1951a01f 100644
--- a/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRangeTest.java
+++ b/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/cluster/locator/TokenRangeTest.java
@@ -21,8 +21,13 @@ package org.apache.cassandra.sidecar.common.server.cluster.locator;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.stream.Stream;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -42,18 +47,6 @@ class TokenRangeTest
.isNotEqualTo(r2);
}
- @Test
- void testFirstToken()
- {
- TokenRange range = new TokenRange(1, 100);
- assertThat(range.firstToken()).isEqualTo(Token.from(2));
- // test the first token refer is the same
- assertThat(range.firstToken()).isSameAs(range.firstToken());
-
- TokenRange emptyRange = new TokenRange(1, 1);
- assertThat(emptyRange.firstToken()).isNull();
- }
-
@Test
void testCreateRangeWithInvalidParams()
{
@@ -114,20 +107,20 @@ class TokenRangeTest
}
@Test
- void testOverlaps()
+ void testIntersects()
{
TokenRange r1 = new TokenRange(3, 5);
TokenRange r2 = new TokenRange(1, 10);
TokenRange r3 = new TokenRange(10, 11);
TokenRange r4 = new TokenRange(4, 11);
- assertThat(r1.overlaps(r2)).isTrue();
- assertThat(r2.overlaps(r1)).isTrue();
- assertThat(r3.overlaps(r4)).isTrue();
- assertThat(r4.overlaps(r3)).isTrue();
- assertThat(r2.overlaps(r4)).isTrue();
- assertThat(r4.overlaps(r2)).isTrue();
- assertThat(r2.overlaps(r3)).isFalse();
- assertThat(r3.overlaps(r2)).isFalse();
+ assertThat(r1.intersects(r2)).isTrue();
+ assertThat(r2.intersects(r1)).isTrue();
+ assertThat(r3.intersects(r4)).isTrue();
+ assertThat(r4.intersects(r3)).isTrue();
+ assertThat(r2.intersects(r4)).isTrue();
+ assertThat(r4.intersects(r2)).isTrue();
+ assertThat(r2.intersects(r3)).isFalse();
+ assertThat(r3.intersects(r2)).isFalse();
}
@Test
@@ -159,6 +152,66 @@ class TokenRangeTest
assertThat(r2.intersection(r4)).isNotEqualTo(new TokenRange(5, 5)); // but not any empty range
}
+ @ParameterizedTest(name = "{index} - {0}: inputLeft={1} inputRight={2} expectedLeft={3} expectedRight={4}")
+ @MethodSource("inputAndExpectedResultAfterSymmetricDiff")
+ void testSymmetricDiff(String testTitle, Set<TokenRange> left, Set<TokenRange> right, Set<TokenRange> expectedLeft, Set<TokenRange> expectedRight)
+ {
+ TokenRange.SymmetricDiffResult symmetricDiffResult = TokenRange.symmetricDiff(left, right);
+ assertThat(symmetricDiffResult.onlyInLeft).isEqualTo(expectedLeft);
+ assertThat(symmetricDiffResult.onlyInRight).isEqualTo(expectedRight);
+
+ // exchange left and right; it is to test the commutative property of diff
+ symmetricDiffResult = TokenRange.symmetricDiff(right, left);
+ assertThat(symmetricDiffResult.onlyInLeft).isEqualTo(expectedRight);
+ assertThat(symmetricDiffResult.onlyInRight).isEqualTo(expectedLeft);
+ }
+
+ public static Stream<Arguments> inputAndExpectedResultAfterSymmetricDiff()
+ {
+ return Stream.of(
+ // inputLeft, inputRight, expectedLeft, expectedRight
+ args("Diff on identical sets",
+ Set.of(r(0, 1000), r(1000, 2000)), // inputLeft
+ Set.of(r(0, 2000)), // inputRight
+ Set.of(), // expectedLeft
+ Set.of()), // expectedRight
+
+ args("Diff on enclosing sets",
+ Set.of(r(0, 1000), r(1000, 2000)), // inputLeft
+ Set.of(r(1000, 2000)), // inputRight
+ Set.of(r(0, 1000)), // expectedLeft
+ Set.of()), // expectedRight
+
+ args("Diff on overlapping sets",
+ Set.of(r(0, 1000), r(1000, 2000)), // inputLeft
+ Set.of(r(500, 1500), r(2000, 2500)), // inputRight
+ Set.of(r(0, 500), r(1500, 2000)), // expectedLeft
+ Set.of(r(2000, 2500))), // expectedRight
+
+ args("Diff on disjoint ranges",
+ Set.of(r(0, 1000)), // inputLeft
+ Set.of(r(2000, 2500)), // inputRight
+ Set.of(r(0, 1000)), // expectedLeft
+ Set.of(r(2000, 2500))), // expectedRight
+
+ args("Diff on overlapping singleton sets",
+ Set.of(r(0, 1000)), // inputLeft
+ Set.of(r(500, 1500)), // inputRight
+ Set.of(r(0, 500)), // expectedLeft
+ Set.of(r(1000, 1500))) // expectedRight
+ );
+ }
+
+ private static TokenRange r(long start, long end)
+ {
+ return new TokenRange(start, end);
+ }
+
+ private static Arguments args(Object... args)
+ {
+ return Arguments.arguments(args);
+ }
+
private com.datastax.driver.core.TokenRange mockRange(long start, long end)
{
com.datastax.driver.core.TokenRange range = mock(com.datastax.driver.core.TokenRange.class);
diff --git a/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/data/RestoreRangeStatusTest.java b/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/data/RestoreRangeStatusTest.java
index 6bb7c374..d4ac35a1 100644
--- a/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/data/RestoreRangeStatusTest.java
+++ b/server-common/src/test/java/org/apache/cassandra/sidecar/common/server/data/RestoreRangeStatusTest.java
@@ -21,9 +21,13 @@ package org.apache.cassandra.sidecar.common.server.data;
import java.util.stream.Stream;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import static org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus.ABORTED;
import static org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus.CREATED;
+import static org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus.DISCARDED;
import static org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus.FAILED;
import static org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus.STAGED;
import static org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus.SUCCEEDED;
@@ -37,33 +41,36 @@ class RestoreRangeStatusTest
{
assertAdvanceTo(CREATED, FAILED);
assertAdvanceTo(CREATED, ABORTED);
+ assertAdvanceTo(CREATED, DISCARDED);
assertAdvanceTo(CREATED, STAGED);
assertAdvanceTo(STAGED, FAILED);
assertAdvanceTo(STAGED, ABORTED);
+ assertAdvanceTo(STAGED, DISCARDED);
assertAdvanceTo(STAGED, SUCCEEDED);
}
- @Test
- void testInvalidStatusAdvancing()
+ @ParameterizedTest(name = "{index}: {0} -> {1}")
+ @MethodSource("invalidStatusAdvancingSource")
+ void testInvalidStatusAdvancing(RestoreRangeStatus from, RestoreRangeStatus to)
{
- String commonErrorMsg = "status can only advance to one of the follow statuses";
+ String commonErrorMsg = from + " status can only advance to one of the follow statuses";
+
+ assertThatThrownBy(() -> from.advanceTo(to)).isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasNoCause()
+ .hasMessageContaining(commonErrorMsg);
+ }
- Stream
- .of(new RestoreRangeStatus[][]
- { // define test cases of invalid status advancing, e.g. it is invalid to advance from EMPTY to STAGED
- { STAGED, CREATED },
- { CREATED, SUCCEEDED },
- { STAGED, STAGED },
- { SUCCEEDED, FAILED },
- { FAILED, SUCCEEDED },
- { FAILED, ABORTED }
- })
- .forEach(testCase -> {
- assertThatThrownBy(() -> testCase[0].advanceTo(testCase[1]))
- .isExactlyInstanceOf(IllegalArgumentException.class)
- .hasNoCause()
- .hasMessageContaining(commonErrorMsg);
- });
+ public static Stream<Arguments> invalidStatusAdvancingSource()
+ {
+ return Stream.of(Arguments.of(STAGED, CREATED),
+ Arguments.of(CREATED, SUCCEEDED),
+ Arguments.of(STAGED, STAGED),
+ Arguments.of(SUCCEEDED, FAILED),
+ Arguments.of(FAILED, SUCCEEDED),
+ Arguments.of(FAILED, ABORTED),
+ Arguments.of(DISCARDED, CREATED),
+ Arguments.of(DISCARDED, STAGED),
+ Arguments.of(DISCARDED, SUCCEEDED));
}
private void assertAdvanceTo(RestoreRangeStatus from, RestoreRangeStatus to)
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/cluster/CassandraAdapterDelegate.java b/server/src/main/java/org/apache/cassandra/sidecar/cluster/CassandraAdapterDelegate.java
index 479a497b..bd959b4d 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/cluster/CassandraAdapterDelegate.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/cluster/CassandraAdapterDelegate.java
@@ -225,7 +225,7 @@ public class CassandraAdapterDelegate implements ICassandraAdapter, Host.StateLi
notifyJmxConnection();
}
- LOGGER.debug("Cassandra version {}", newNodeSettings.releaseVersion());
+ LOGGER.trace("Cassandra version {}", newNodeSettings.releaseVersion());
}
catch (RuntimeException e)
{
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/cluster/locator/CachedLocalTokenRanges.java b/server/src/main/java/org/apache/cassandra/sidecar/cluster/locator/CachedLocalTokenRanges.java
index fadc2eca..d1562e57 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/cluster/locator/CachedLocalTokenRanges.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/cluster/locator/CachedLocalTokenRanges.java
@@ -77,8 +77,13 @@ public class CachedLocalTokenRanges implements LocalTokenRangesProvider
this.localInstancesCache = null;
}
+ /**
+ * {@inheritDoc}
+ *
+ * @param forceRefresh the parameter is ignored. This implementation always serves from cache first and manages reload internally
+ */
@Override
- public Map<Integer, Set<TokenRange>> localTokenRanges(String keyspace)
+ public Map<Integer, Set<TokenRange>> localTokenRanges(String keyspace, boolean forceRefresh)
{
List<InstanceMetadata> localInstances = instancesMetadata.instances();
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/cluster/locator/LocalTokenRangesProvider.java b/server/src/main/java/org/apache/cassandra/sidecar/cluster/locator/LocalTokenRangesProvider.java
index ddc7da63..783ddae5 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/cluster/locator/LocalTokenRangesProvider.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/cluster/locator/LocalTokenRangesProvider.java
@@ -28,6 +28,17 @@ import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
*/
public interface LocalTokenRangesProvider
{
+ /**
+ * Similar to {@link #localTokenRanges(String, boolean)}, but always refresh forcibly.
+ *
+ * @param keyspace keyspace to determine replication
+ * @return token ranges of the local Cassandra instances or an empty map of nothing is found
+ */
+ default Map<Integer, Set<TokenRange>> localTokenRanges(String keyspace)
+ {
+ return localTokenRanges(keyspace, true);
+ }
+
/**
* Calculate the token ranges owned and replicated to the local Cassandra instance(s).
* When Sidecar is paired with multiple Cassandra instance, the ranges of each Cassandra instance is captured
@@ -36,7 +47,8 @@ public interface LocalTokenRangesProvider
* When Sidecar is paired with a single Cassandra instance, the result map has a single entry.
*
* @param keyspace keyspace to determine replication
+ * @param forceRefresh whether force refreshing is requested or not
* @return token ranges of the local Cassandra instances or an empty map of nothing is found
*/
- Map<Integer, Set<TokenRange>> localTokenRanges(String keyspace);
+ Map<Integer, Set<TokenRange>> localTokenRanges(String keyspace, boolean forceRefresh);
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/PeriodicTaskConfiguration.java b/server/src/main/java/org/apache/cassandra/sidecar/config/PeriodicTaskConfiguration.java
index 8b99603a..a0f51098 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/config/PeriodicTaskConfiguration.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/config/PeriodicTaskConfiguration.java
@@ -19,6 +19,7 @@
package org.apache.cassandra.sidecar.config;
import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.jetbrains.annotations.NotNull;
/**
* Configuration relevant for {@link org.apache.cassandra.sidecar.tasks.PeriodicTask}s
@@ -33,11 +34,13 @@ public interface PeriodicTaskConfiguration
/**
* @return the initial delay for the first execution of this task after being scheduled or rescheduled
*/
+ @NotNull
MillisecondBoundConfiguration initialDelay();
/**
* @return how often this task will execute after the previous task has completed the {@link io.vertx.core.Promise}
* of the execution
*/
+ @NotNull
MillisecondBoundConfiguration executeInterval();
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/PeriodicTaskConfigurationImpl.java b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/PeriodicTaskConfigurationImpl.java
index 2897759b..78d115cb 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/PeriodicTaskConfigurationImpl.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/PeriodicTaskConfigurationImpl.java
@@ -43,6 +43,9 @@ public class PeriodicTaskConfigurationImpl implements PeriodicTaskConfiguration
private MillisecondBoundConfiguration initialDelay;
private MillisecondBoundConfiguration executeInterval;
+ /**
+ * Default constructor for jackson deserialization and allows to set the fields, including the aliased ones.
+ */
public PeriodicTaskConfigurationImpl()
{
this.enabled = DEFAULT_ENABLED;
@@ -144,7 +147,7 @@ public class PeriodicTaskConfigurationImpl implements PeriodicTaskConfiguration
public void setExecuteIntervalMillis(long executeIntervalMillis)
{
LOGGER.warn("'execute_interval_millis', 'poll_freq_millis', and 'poll_interval_millis' are deprecated, " +
- "use 'execute_interval' instead");
+ "use 'execute_interval' instead.");
setExecuteInterval(new MillisecondBoundConfiguration(executeIntervalMillis, TimeUnit.MILLISECONDS));
}
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreRange.java b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreRange.java
index debb6e3b..dcf21813 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreRange.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreRange.java
@@ -22,6 +22,7 @@ import java.math.BigInteger;
import java.net.InetSocketAddress;
import java.nio.file.Path;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
@@ -33,6 +34,7 @@ import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
import org.apache.cassandra.sidecar.cluster.locator.LocalTokenRangesProvider;
import org.apache.cassandra.sidecar.common.DataObjectBuilder;
import org.apache.cassandra.sidecar.common.response.data.RestoreRangeJson;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus;
import org.apache.cassandra.sidecar.common.server.utils.StringUtils;
import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
@@ -80,6 +82,9 @@ import org.jetbrains.annotations.VisibleForTesting;
*/
public class RestoreRange
{
+ public static final Comparator<RestoreRange> TOKEN_BASED_NATURAL_ORDER = Comparator.comparing(RestoreRange::startToken)
+ .thenComparing(RestoreRange::endToken);
+
// @NotNull fields are persisted
@NotNull
private final UUID jobId;
@@ -92,9 +97,7 @@ public class RestoreRange
@NotNull
private final String sliceKey;
@NotNull
- private final BigInteger startToken;
- @NotNull
- private final BigInteger endToken;
+ private final TokenRange tokenRange;
@NotNull
private final Map<String, RestoreRangeStatus> statusByReplica;
@@ -118,6 +121,7 @@ public class RestoreRange
private boolean hasImported = false;
private int downloadAttempt = 0;
private volatile boolean isCancelled = false;
+ private volatile boolean discarded = false;
public static RestoreRange from(Row row)
{
@@ -142,8 +146,7 @@ public class RestoreRange
{
this.jobId = builder.jobId;
this.bucketId = builder.bucketId;
- this.startToken = builder.startToken;
- this.endToken = builder.endToken;
+ this.tokenRange = new TokenRange(builder.startToken, builder.endToken);
this.source = builder.sourceSlice;
this.sliceId = builder.sliceId;
this.sliceKey = builder.sliceKey;
@@ -154,6 +157,7 @@ public class RestoreRange
this.owner = builder.owner;
this.statusByReplica = new HashMap<>(builder.statusByReplica);
this.tracker = builder.tracker;
+ this.discarded = builder.discarded;
}
public Builder unbuild()
@@ -163,13 +167,13 @@ public class RestoreRange
public RestoreRangeJson toJson()
{
- return new RestoreRangeJson(sliceId, bucketId, sliceBucket, sliceKey, startToken, endToken);
+ return new RestoreRangeJson(sliceId, bucketId, sliceBucket, sliceKey, tokenRange.startAsBigInt(), tokenRange.endAsBigInt());
}
@Override
public int hashCode()
{
- return Objects.hash(startToken, endToken, jobId, bucketId, sliceId);
+ return Objects.hash(tokenRange, jobId, bucketId, sliceId);
}
@Override
@@ -184,12 +188,25 @@ public class RestoreRange
RestoreRange that = (RestoreRange) obj;
// Note: destinationPathInStaging and owner are not included as they are 'transient'.
// Mutable states are not included, e.g. status_by_replicas.
- return Objects.equals(this.startToken, that.startToken)
- && Objects.equals(this.endToken, that.endToken)
+ return Objects.equals(this.tokenRange, that.tokenRange)
&& Objects.equals(this.jobId, that.jobId)
&& Objects.equals(this.bucketId, that.bucketId)
&& Objects.equals(this.sliceId, that.sliceId);
}
+
+ @Override
+ public String toString()
+ {
+ return "RestoreRange{" +
+ "jobId=" + jobId +
+ ", sliceId='" + sliceId + '\'' +
+ ", tokenRange=" + tokenRange +
+ ", statusByReplica=" + statusByReplica +
+ ", sliceKey='" + sliceKey + '\'' +
+ ", sliceBucket='" + sliceBucket + '\'' +
+ '}';
+ }
+
// -- INTERNAL FLOW CONTROL METHODS --
/**
@@ -254,6 +271,21 @@ public class RestoreRange
isCancelled = true;
}
+ /**
+ * Mark the restore range as discarded for this instance, then cancel it
+ */
+ public void discard()
+ {
+ discarded = true;
+ updateRangeStatusForInstance(owner(), RestoreRangeStatus.DISCARDED);
+ cancel();
+ }
+
+ public boolean isDiscarded()
+ {
+ return discarded;
+ }
+
/**
* @return {@link RestoreRangeTask} that defines the steps to download and import data into Cassandra
*/
@@ -377,12 +409,17 @@ public class RestoreRange
public BigInteger startToken()
{
- return this.startToken;
+ return tokenRange.startAsBigInt();
}
public BigInteger endToken()
{
- return this.endToken;
+ return tokenRange.endAsBigInt();
+ }
+
+ public TokenRange tokenRange()
+ {
+ return tokenRange;
}
public Map<String, RestoreRangeStatus> statusByReplica()
@@ -529,6 +566,7 @@ public class RestoreRange
private String uploadId;
private Map<String, RestoreRangeStatus> statusByReplica = Collections.emptyMap();
private RestoreJobProgressTracker tracker = null;
+ private boolean discarded;
private Builder()
{
@@ -545,10 +583,11 @@ public class RestoreRange
this.stageDirectory = range.stageDirectory;
this.uploadId = range.uploadId;
this.owner = range.owner;
- this.startToken = range.startToken;
- this.endToken = range.endToken;
- this.statusByReplica = Collections.unmodifiableMap(range.statusByReplica);
+ this.startToken = range.tokenRange.startAsBigInt();
+ this.endToken = range.tokenRange.endAsBigInt();
+ this.statusByReplica = new HashMap<>(range.statusByReplica);
this.tracker = range.tracker;
+ this.discarded = range.discarded;
}
public Builder jobId(UUID jobId)
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreRangeDatabaseAccessor.java b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreRangeDatabaseAccessor.java
index 516611a4..9f06ef2b 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreRangeDatabaseAccessor.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreRangeDatabaseAccessor.java
@@ -22,6 +22,9 @@ import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
@@ -38,6 +41,8 @@ import org.apache.cassandra.sidecar.db.schema.SidecarSchema;
@Singleton
public class RestoreRangeDatabaseAccessor extends DatabaseAccessor<RestoreRangesSchema>
{
+ private static final Logger LOGGER = LoggerFactory.getLogger(RestoreRangeDatabaseAccessor.class);
+
private final SidecarSchema sidecarSchema;
@Inject
@@ -53,16 +58,17 @@ public class RestoreRangeDatabaseAccessor extends DatabaseAccessor<RestoreRanges
{
sidecarSchema.ensureInitialized();
- BoundStatement statement = tableSchema.insert()
- .bind(range.jobId(),
- range.bucketId(),
- range.startToken(),
- range.endToken(),
- range.sliceId(),
+ BoundStatement statement = tableSchema.createRange()
+ .bind(range.sliceId(),
range.sliceBucket(),
range.sliceKey(),
- range.statusTextByReplica());
+ range.statusTextByReplica(),
+ range.jobId(),
+ range.bucketId(),
+ range.startToken(),
+ range.endToken());
execute(statement);
+ LOGGER.debug("Created range={}", range);
return range;
}
@@ -77,6 +83,7 @@ public class RestoreRangeDatabaseAccessor extends DatabaseAccessor<RestoreRanges
range.startToken(),
range.endToken());
execute(statement);
+ LOGGER.debug("Updated range={}", range);
return range;
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreSlice.java b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreSlice.java
index a62653ab..5caa4be4 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreSlice.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreSlice.java
@@ -19,32 +19,21 @@
package org.apache.cassandra.sidecar.db;
import java.math.BigInteger;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
import java.util.Objects;
import java.util.UUID;
-import java.util.stream.Collectors;
-import javax.annotation.Nullable;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.datastax.driver.core.Row;
import org.apache.cassandra.sidecar.common.DataObjectBuilder;
import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
-import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
-import org.apache.cassandra.sidecar.common.server.cluster.locator.Token;
import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.jetbrains.annotations.NotNull;
/**
* Data object that contains all values that matter to the restore job slice.
*/
public class RestoreSlice
{
- private static final Logger LOGGER = LoggerFactory.getLogger(RestoreSlice.class);
-
private final UUID jobId;
private final String keyspace;
private final String table;
@@ -53,8 +42,8 @@ public class RestoreSlice
private final String bucket;
private final String key;
private final String checksum; // etag
- private final BigInteger startToken;
- private final BigInteger endToken;
+ private final BigInteger startToken; // exclusive
+ private final BigInteger endToken; // inclusive
private final long creationTimeNanos;
private final long compressedSize;
private final long uncompressedSize;
@@ -118,69 +107,40 @@ public class RestoreSlice
}
/**
- * Splits the slice based on the topology changes.
- * <p></p>
- * When the current topology has pending ranges, it results into the slice S being split into
- * multiple sub-ranges, say, s1 and s2. The updater splits, and persists the new slices to database.
- * The original slice S is then deleted.
- * The new slices still reference to the same s3 object, i.e. {@code <bucket/key/checksum>}
+ * Trim the slice based on the reference local token range.
+ *
+ * <p>The range of the slice might not be entirely enclosed by the localTokenRange.
+ * In such case, the slice is trimmed to align with the localTokenRange. For example,
+ * the slice covers range {@code (1, 100])} and the localTokenRange covers {@code (50, 90]}.
+ * The slice is trimmed to match with the localTokenRange, updating the range to {@code (50, 90]}.
+ * <p>The trimmed slice still reference to the same s3 object, i.e. {@code <bucket/key/checksum>}
*
- * @param topology topology to guide the splitting
- * @return a list of slices of the sub-ranges
+ * @param localTokenRange local token range
+ * @return a restore slice that might be trimmed
*/
- public List<RestoreSlice> splitMaybe(@Nullable TokenRangeReplicasResponse topology)
+ public RestoreSlice trimMaybe(@NotNull TokenRange localTokenRange)
{
- if (topology == null)
- {
- return Collections.singletonList(this);
- }
-
- List<RestoreSlice> splits = new ArrayList<>();
TokenRange sliceRange = new TokenRange(startToken(), endToken());
- for (TokenRangeReplicasResponse.ReplicaInfo replicaInfo : topology.writeReplicas())
+ if (localTokenRange.encloses(sliceRange))
{
- TokenRange range = new TokenRange(Token.from(replicaInfo.start()), Token.from(replicaInfo.end()));
- if (range.encloses(sliceRange))
- {
- return Collections.singletonList(this);
- }
- else if (range.overlaps(sliceRange))
- {
- TokenRange intersection = range.intersection(sliceRange);
- // Adjust the slice range to match with the write replica and persist.
- // The object location remains the same as sidecar need to download the same object.
- // It just needs a narrower range of data within the slice
- splits.add(unbuild()
- .startToken(intersection.start().toBigInteger())
- .endToken(intersection.end().toBigInteger())
- .build());
- }
-
- if (range.largerThan(sliceRange))
- {
- // all following ranges are larger the original range; exit the iteration early.
- break;
- }
+ return this;
}
- if (splits.isEmpty())
- {
- throw new IllegalStateException("Token range of the slice is not found in the write replicas. " +
- "slice range: " + formatRange(startToken, endToken));
- }
- else
+ if (localTokenRange.intersects(sliceRange))
{
- LOGGER.info("The slice is split. splitsCount={} splits={}",
- splits.size(), splits.stream().map(s -> formatRange(s.startToken, s.endToken))
- .collect(Collectors.toList()));
+ TokenRange intersection = localTokenRange.intersection(sliceRange);
+ // Adjust the slice range to match with localTokenRange
+ // The object location remains the same as sidecar need to download the same object.
+ // It only narrows the range of data within the slice
+ return unbuild()
+ .startToken(intersection.start().toBigInteger())
+ .endToken(intersection.end().toBigInteger())
+ .build();
}
- return splits;
- }
-
- private String formatRange(BigInteger start, BigInteger end)
- {
- return "(" + start + ", " + end + ']';
+ throw new IllegalStateException("Token range of the slice does not intersect with the local token range. " +
+ "slice_range: " + sliceRange +
+ ", local_range: " + localTokenRange);
}
// -- (self-explanatory) GETTERS --
@@ -283,8 +243,8 @@ public class RestoreSlice
private String bucket;
private String key;
private String checksum; // etag
- private BigInteger startToken;
- private BigInteger endToken;
+ private BigInteger startToken; // exclusive
+ private BigInteger endToken; // inclusive
private long compressedSize;
private long uncompressedSize;
@@ -388,7 +348,7 @@ public class RestoreSlice
b.bucket = payload.bucket();
b.key = payload.key();
b.checksum = payload.checksum();
- b.startToken = payload.startToken();
+ b.startToken = payload.firstToken().subtract(BigInteger.ONE);
b.endToken = payload.endToken();
b.compressedSize = payload.compressedSizeOrZero();
b.uncompressedSize = payload.uncompressedSizeOrZero();
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreSliceDatabaseAccessor.java b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreSliceDatabaseAccessor.java
index 640bd427..f497e60b 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreSliceDatabaseAccessor.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreSliceDatabaseAccessor.java
@@ -27,9 +27,7 @@ import com.datastax.driver.core.Row;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
-import org.apache.cassandra.sidecar.common.server.cluster.locator.Token;
import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
-import org.apache.cassandra.sidecar.common.utils.Preconditions;
import org.apache.cassandra.sidecar.db.schema.RestoreSlicesSchema;
import org.apache.cassandra.sidecar.db.schema.SidecarSchema;
@@ -80,13 +78,10 @@ public class RestoreSliceDatabaseAccessor extends DatabaseAccessor<RestoreSlices
public List<RestoreSlice> selectByJobByBucketByTokenRange(RestoreJob restoreJob, short bucketId, TokenRange range)
{
sidecarSchema.ensureInitialized();
- Token firstToken = range.firstToken();
- Preconditions.checkArgument(firstToken != null, "range cannot be empty");
-
BoundStatement statement = tableSchema.findAllByTokenRange()
.bind(restoreJob.jobId,
bucketId,
- firstToken.toBigInteger(),
+ range.start().toBigInteger(),
range.end().toBigInteger());
ResultSet result = execute(statement);
List<RestoreSlice> slices = new ArrayList<>();
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreRangesSchema.java b/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreRangesSchema.java
index 5d552fa7..7ad57b63 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreRangesSchema.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreRangesSchema.java
@@ -36,7 +36,7 @@ public class RestoreRangesSchema extends TableSchema implements ExecuteOnCluster
private final SchemaKeyspaceConfiguration keyspaceConfig;
private final SecondBoundConfiguration tableTtl;
- private PreparedStatement insert;
+ private PreparedStatement createRange;
private PreparedStatement findAll;
private PreparedStatement update;
@@ -55,7 +55,7 @@ public class RestoreRangesSchema extends TableSchema implements ExecuteOnCluster
@Override
protected void prepareStatements(@NotNull Session session)
{
- insert = prepare(insert, session, CqlLiterals.insert(keyspaceConfig));
+ createRange = prepare(createRange, session, CqlLiterals.createRange(keyspaceConfig));
findAll = prepare(findAll, session, CqlLiterals.findAll(keyspaceConfig));
update = prepare(update, session, CqlLiterals.update(keyspaceConfig));
}
@@ -72,20 +72,20 @@ public class RestoreRangesSchema extends TableSchema implements ExecuteOnCluster
return String.format("CREATE TABLE IF NOT EXISTS %s.%s (" +
" job_id timeuuid," +
" bucket_id smallint," + // same bucket_id as in the slice row
- " start_token varint," +
- " end_token varint," +
+ " start_token varint," + // exclusive
+ " end_token varint," + // inclusive
" slice_id text," +
" slice_bucket text," +
" slice_key text," +
- " status_by_replica map<text, text>," +
+ " status_by_replica map<text, text>," + // key is the replica; value is the status
" PRIMARY KEY ((job_id, bucket_id), start_token, end_token)" +
") WITH default_time_to_live = %s",
keyspaceConfig.keyspace(), TABLE_NAME, tableTtl.toSeconds());
}
- public PreparedStatement insert()
+ public PreparedStatement createRange()
{
- return insert;
+ return createRange;
}
public PreparedStatement findAll()
@@ -100,18 +100,12 @@ public class RestoreRangesSchema extends TableSchema implements ExecuteOnCluster
private static class CqlLiterals
{
- static String insert(SchemaKeyspaceConfiguration config)
+ static String createRange(SchemaKeyspaceConfiguration config)
{
- return withTable("INSERT INTO %s.%s (" +
- " job_id," +
- " bucket_id," +
- " start_token," +
- " end_token," +
- " slice_id," +
- " slice_bucket," +
- " slice_key," +
- " status_by_replica" +
- ") VALUES (?, ?, ?, ?, ?, ?, ?, ?)", config);
+ // Use update to create range to avoid over-writing status_by_replica map
+ return withTable("UPDATE %s.%s " +
+ "SET slice_id = ?, slice_bucket = ?, slice_key = ?, status_by_replica = status_by_replica + ? " +
+ "WHERE job_id = ? AND bucket_id = ? AND start_token = ? AND end_token = ?", config);
}
// ALLOW FILTERING within the same partition should have minimum impact on read performance.
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreSlicesSchema.java b/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreSlicesSchema.java
index bb56ff9e..095ea413 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreSlicesSchema.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreSlicesSchema.java
@@ -114,15 +114,15 @@ public class RestoreSlicesSchema extends TableSchema implements ExecuteOnCluster
}
// ALLOW FILTERING within the same partition should have minimum impact on read performance.
- // To locate all qualified (intersecting) ranges, for Range [T1, T2],
- // the conditions is `start_token <= T2 AND end_token >= T1`
+ // To locate all qualified (intersecting) ranges, for the bind input range (T1, T2] and
+ // the range (start_token, end_token] to intersect, `start_token < T2 AND end_token > T1`
static String findAllByTokenRange(SchemaKeyspaceConfiguration config)
{
return withTable("SELECT job_id, bucket_id, slice_id, bucket, key, checksum, " +
"start_token, end_token, compressed_size, uncompressed_size " +
"FROM %s.%s " +
"WHERE job_id = ? AND bucket_id = ? AND " +
- "end_token >= ? AND start_token <= ? ALLOW FILTERING", config);
+ "end_token > ? AND start_token < ? ALLOW FILTERING", config);
}
private static String withTable(String format, SchemaKeyspaceConfiguration config)
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/db/schema/SystemAuthSchema.java b/server/src/main/java/org/apache/cassandra/sidecar/db/schema/SystemAuthSchema.java
index 274b2696..d83ab1f9 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/db/schema/SystemAuthSchema.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/schema/SystemAuthSchema.java
@@ -56,7 +56,7 @@ public class SystemAuthSchema extends CassandraSystemTableSchema
// identity_to_role table exists in Cassandra versions starting 5.x
if (keyspaceMetadata == null || keyspaceMetadata.getTable(IDENTITY_TO_ROLE_TABLE) == null)
{
- logger.info("Auth table does not exist. Skip preparing. table={}.{}", keyspaceName(), IDENTITY_TO_ROLE_TABLE);
+ logger.info("system_auth.identity_to_role does not exist. Skip preparing. table={}.{}", keyspaceName(), IDENTITY_TO_ROLE_TABLE);
return;
}
roleFromIdentity = prepare(roleFromIdentity, session, "SELECT role FROM system_auth.identity_to_role WHERE identity = ?");
@@ -103,8 +103,7 @@ public class SystemAuthSchema extends CassandraSystemTableSchema
{
if (roleFromIdentity == null || allRolesAndIdentities == null)
{
- throw new SchemaUnavailableException(String.format("Table %s.%s does not exist",
- keyspaceName(), IDENTITY_TO_ROLE_TABLE));
+ throw new SchemaUnavailableException(keyspaceName(), IDENTITY_TO_ROLE_TABLE);
}
}
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyLevelChecker.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyChecker.java
similarity index 65%
rename from server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyLevelChecker.java
rename to server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyChecker.java
index 637e97d9..477648ef 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyLevelChecker.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyChecker.java
@@ -26,6 +26,10 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import com.google.common.collect.Range;
+import com.google.common.collect.RangeMap;
+import com.google.common.collect.TreeRangeMap;
+import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,12 +64,12 @@ import static java.util.stream.Collectors.mapping;
import static java.util.stream.Collectors.toSet;
/**
- * Checks restore job with the configured consistency level
+ * Checks restore job import consistency with the configured consistency level
*/
@Singleton
-public class RestoreJobConsistencyLevelChecker
+public class RestoreJobConsistencyChecker
{
- private static final Logger LOGGER = LoggerFactory.getLogger(RestoreJobConsistencyLevelChecker.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(RestoreJobConsistencyChecker.class);
private final RingTopologyRefresher ringTopologyRefresher;
private final RestoreJobDiscoverer restoreJobDiscoverer;
@@ -75,11 +79,11 @@ public class RestoreJobConsistencyLevelChecker
private volatile boolean firstTimeSinceImportReady = true;
@Inject
- public RestoreJobConsistencyLevelChecker(RingTopologyRefresher ringTopologyRefresher,
- RestoreJobDiscoverer restoreJobDiscoverer,
- RestoreRangeDatabaseAccessor rangeDatabaseAccessor,
- ExecutorPools executorPools,
- SidecarMetrics sidecarMetrics)
+ public RestoreJobConsistencyChecker(RingTopologyRefresher ringTopologyRefresher,
+ RestoreJobDiscoverer restoreJobDiscoverer,
+ RestoreRangeDatabaseAccessor rangeDatabaseAccessor,
+ ExecutorPools executorPools,
+ SidecarMetrics sidecarMetrics)
{
this.ringTopologyRefresher = ringTopologyRefresher;
this.restoreJobDiscoverer = restoreJobDiscoverer;
@@ -146,45 +150,103 @@ public class RestoreJobConsistencyLevelChecker
return false;
}
+ @SuppressWarnings("UnstableApiUsage")
private static void concludeRanges(List<RestoreRange> ranges,
TokenRangeReplicasResponse topology,
ConsistencyVerifier verifier,
RestoreRangeStatus successCriteria,
RestoreJobProgressCollector collector)
{
- for (RestoreRange range : ranges)
+ RangeMap<Token, InstanceSetByDc> replicasPerRange = populateReplicas(topology);
+ Map<Range<Token>, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> statusPerRange = populateStatusByReplica(ranges);
+
+ for (Map.Entry<Range<Token>, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> entry : statusPerRange.entrySet())
{
if (!collector.canCollectMore())
{
return;
}
- ConsistencyVerificationResult res = concludeOneRange(topology, verifier, successCriteria, range);
- collector.collect(range, res);
+ Range<Token> tokenRange = entry.getKey();
+ Map<String, RestoreRangeStatus> status = entry.getValue().getLeft();
+ RestoreRange relevantRestoreRange = entry.getValue().getRight();
+ ConsistencyVerificationResult res = concludeOneRange(replicasPerRange, verifier, successCriteria, tokenRange, status);
+ collector.collect(relevantRestoreRange, res);
}
}
+ @SuppressWarnings("UnstableApiUsage")
+ static RangeMap<Token, InstanceSetByDc> populateReplicas(TokenRangeReplicasResponse topology)
+ {
+ RangeMap<Token, InstanceSetByDc> replicasPerRange = TreeRangeMap.create();
+ for (TokenRangeReplicasResponse.ReplicaInfo replicaInfo : topology.writeReplicas())
+ {
+ TokenRange tokenRange = new TokenRange(Token.from(replicaInfo.start()), Token.from(replicaInfo.end()));
+ Map<String, List<String>> replicasByDc = replicaInfo.replicasByDatacenter();
+ Map<String, Set<String>> mapping = new HashMap<>(replicasByDc.size());
+ replicasByDc.forEach((dc, instances) -> mapping.put(dc, new HashSet<>(instances)));
+ InstanceSetByDc instanceSetByDc = new InstanceSetByDc(mapping);
+ replicasPerRange.put(tokenRange.range, instanceSetByDc);
+ }
+ return replicasPerRange;
+ }
+
+ @SuppressWarnings("UnstableApiUsage")
+ static Map<Range<Token>, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> populateStatusByReplica(List<RestoreRange> ranges)
+ {
+ RangeMap<Token, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> rangeMap = TreeRangeMap.create();
+ rangeMap.put(Range.all(), Pair.of(Collections.emptyMap(), null));
+ for (RestoreRange restoreRange : ranges)
+ {
+ Range<Token> tokenRange = restoreRange.tokenRange().range;
+ RangeMap<Token, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> sub = rangeMap.subRangeMap(tokenRange);
+ Map<Range<Token>, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> map = sub.asMapOfRanges();
+ if (map.isEmpty())
+ {
+ rangeMap.put(tokenRange, Pair.of(restoreRange.statusByReplica(), restoreRange));
+ }
+ else
+ {
+ // If subrange map is non-empty, i.e. RestoreRanges has overlapping, the overlapping token range should belong to the same RestoreSlice.
+ // Because RestoreRange is derived from RestoreSlice and RestoreSlices do not overlap.
+ RangeMap<Token, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> updated = TreeRangeMap.create();
+ map.forEach((key, value) -> {
+ Map<String, RestoreRangeStatus> statusMap = new HashMap<>(value.getKey());
+ statusMap.putAll(restoreRange.statusByReplica());
+ updated.put(key, Pair.of(statusMap, restoreRange));
+ });
+ rangeMap.putAll(updated);
+ }
+ }
+ Map<Range<Token>, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> result = rangeMap.asMapOfRanges();
+ result.entrySet().removeIf(e -> e.getValue().getValue() == null);
+ return result;
+ }
+
/**
* Examine a range and all its replica should be in the expected status, i.e. {@param successCriteria}.
* If enough replicas are in the expected status, the conclusion can be made that the range has satisfied.
- * If enough replicas are {@link RestoreRangeStatus.FAILED}, it concludes that the range has failed.
+ * If enough replicas are {@link RestoreRangeStatus#FAILED}, it concludes that the range has failed.
* Otherwise, no conclusion is made and the range is pending.
*
- * @param topology current cluster topology
+ * @param replicasByRange replica set of each token range
* @param verifier check whether the replicas status can satisfy the consistency level
* @param successCriteria the expected {@link RestoreRangeStatus} for replicas
* @param range range to check
+ * @param statusByReplica restore progress of each instance
* @return result of the consistency verification
*/
- private static ConsistencyVerificationResult concludeOneRange(TokenRangeReplicasResponse topology,
+ @SuppressWarnings("UnstableApiUsage")
+ private static ConsistencyVerificationResult concludeOneRange(RangeMap<Token, InstanceSetByDc> replicasByRange,
ConsistencyVerifier verifier,
RestoreRangeStatus successCriteria,
- RestoreRange range)
+ Range<Token> range,
+ Map<String, RestoreRangeStatus> statusByReplica)
{
- Map<RestoreRangeStatus, Set<String>> groupByStatus = groupReplicaByStatus(range.statusByReplica());
+ Map<RestoreRangeStatus, Set<String>> groupByStatus = groupReplicaByStatus(statusByReplica);
Set<String> succeeded = groupByStatus.getOrDefault(successCriteria, Collections.emptySet());
Set<String> failed = groupByStatus.getOrDefault(RestoreRangeStatus.FAILED, Collections.emptySet());
- InstanceSetByDc replicaSet = replicaSetForRange(range, topology);
+ InstanceSetByDc replicaSet = replicaSetForRange(range, replicasByRange);
if (replicaSet == null) // cannot proceed to verify yet. Return pending
{
return ConsistencyVerificationResult.PENDING;
@@ -212,36 +274,32 @@ public class RestoreJobConsistencyLevelChecker
/**
* Find the replica set for the token range. Returns null if no complete match can be found or topology has changed.
+ * The input tokenRange should be fully enclosed by one of the range in the replicasByRange
*/
- private static @Nullable InstanceSetByDc replicaSetForRange(RestoreRange range, TokenRangeReplicasResponse topology)
+ @SuppressWarnings("UnstableApiUsage")
+ private static @Nullable InstanceSetByDc replicaSetForRange(Range<Token> tokenRange, RangeMap<Token, InstanceSetByDc> replicasByRange)
{
- TokenRange restoreRange = new TokenRange(range.startToken(), range.endToken());
- for (TokenRangeReplicasResponse.ReplicaInfo replicaInfo : topology.writeReplicas())
+ Map<Range<Token>, InstanceSetByDc> subRange = replicasByRange.subRangeMap(tokenRange).asMapOfRanges();
+ if (subRange.size() == 1)
{
- TokenRange tokenRange = new TokenRange(Token.from(replicaInfo.start()), Token.from(replicaInfo.end()));
- if (tokenRange.encloses(restoreRange))
+ // finding one range does not necessarily mean the found range fully encloses the token range
+ // subRangeMap could contain the partially overlapping range
+ Map.Entry<Range<Token>, InstanceSetByDc> foundEntry = subRange.entrySet().iterator().next();
+ if (foundEntry.getKey().encloses(tokenRange))
{
- Map<String, List<String>> replicasByDc = replicaInfo.replicasByDatacenter();
- Map<String, Set<String>> mapping = new HashMap<>(replicasByDc.size());
- replicasByDc.forEach((k, instances) -> mapping.put(k, new HashSet<>(instances)));
- return new InstanceSetByDc(mapping);
+ return foundEntry.getValue();
}
- else if (tokenRange.overlaps(restoreRange))
+ else
{
- LOGGER.info("Topology change detected");
+ LOGGER.info("Topology is changed");
return null;
}
-
- if (tokenRange.largerThan(restoreRange))
- {
- // all following ranges are larger the original range; exit the iteration early.
- break;
- }
}
-
- LOGGER.warn("Unable to find a complete match for range. startToken={} endToken={}",
- range.startToken(), range.endToken());
- return null;
+ else
+ {
+ LOGGER.warn("Unable to find a single match for range. tokenRange={} matchingRanges={}", tokenRange, subRange);
+ return null;
+ }
}
private static long sliceCountFromRanges(List<RestoreRange> ranges)
@@ -255,12 +313,12 @@ public class RestoreJobConsistencyLevelChecker
RestoreRangeStatus successCriteria,
RestoreRange range)
{
- return concludeOneRange(topology, verifier, successCriteria, range);
+ return concludeOneRange(populateReplicas(topology), verifier, successCriteria, range.tokenRange().range, range.statusByReplica());
}
@VisibleForTesting
static InstanceSetByDc replicaSetForRangeUnsafe(RestoreRange range, TokenRangeReplicasResponse topology)
{
- return replicaSetForRange(range, topology);
+ return replicaSetForRange(range.tokenRange().range, populateReplicas(topology));
}
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
index 1e98ee9d..0243c9d2 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
@@ -19,17 +19,19 @@
package org.apache.cassandra.sidecar.restore;
import java.nio.file.Paths;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,12 +46,16 @@ import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
+import org.apache.cassandra.sidecar.common.utils.Preconditions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
import org.apache.cassandra.sidecar.config.RestoreJobConfiguration;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
import org.apache.cassandra.sidecar.db.RestoreJob;
import org.apache.cassandra.sidecar.db.RestoreJobDatabaseAccessor;
import org.apache.cassandra.sidecar.db.RestoreRange;
import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
+import org.apache.cassandra.sidecar.db.RestoreSlice;
import org.apache.cassandra.sidecar.db.RestoreSliceDatabaseAccessor;
import org.apache.cassandra.sidecar.db.schema.SidecarSchema;
import org.apache.cassandra.sidecar.exceptions.RestoreJobFatalException;
@@ -64,7 +70,7 @@ import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
* {@link RestoreJobDiscoverer} handles background restore job discovery and handling it according to job status
*/
@Singleton
-public class RestoreJobDiscoverer implements PeriodicTask
+public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeListener
{
private static final Logger LOGGER = LoggerFactory.getLogger(RestoreJobDiscoverer.class);
@@ -80,6 +86,7 @@ public class RestoreJobDiscoverer implements PeriodicTask
private final JobIdsByDay jobIdsByDay;
private final RingTopologyRefresher ringTopologyRefresher;
private final AtomicBoolean isExecuting = new AtomicBoolean(false);
+ private final TaskExecutorPool executorPool;
private int inflightJobsCount = 0;
private int jobDiscoveryRecencyDays;
private PeriodicTaskExecutor periodicTaskExecutor;
@@ -91,9 +98,9 @@ public class RestoreJobDiscoverer implements PeriodicTask
RestoreSliceDatabaseAccessor restoreSliceDatabaseAccessor,
RestoreRangeDatabaseAccessor restoreRangeDatabaseAccessor,
Provider<RestoreJobManagerGroup> restoreJobManagerGroupProvider,
- LocalTokenRangesProvider cachedLocalTokenRanges,
InstanceMetadataFetcher instanceMetadataFetcher,
RingTopologyRefresher ringTopologyRefresher,
+ ExecutorPools executorPools,
SidecarMetrics metrics)
{
this(config.restoreJobConfiguration(),
@@ -102,9 +109,9 @@ public class RestoreJobDiscoverer implements PeriodicTask
restoreSliceDatabaseAccessor,
restoreRangeDatabaseAccessor,
restoreJobManagerGroupProvider,
- cachedLocalTokenRanges,
instanceMetadataFetcher,
ringTopologyRefresher,
+ executorPools,
metrics);
}
@@ -115,9 +122,9 @@ public class RestoreJobDiscoverer implements PeriodicTask
RestoreSliceDatabaseAccessor restoreSliceDatabaseAccessor,
RestoreRangeDatabaseAccessor restoreRangeDatabaseAccessor,
Provider<RestoreJobManagerGroup> restoreJobManagerGroupProvider,
- LocalTokenRangesProvider cachedLocalTokenRanges,
InstanceMetadataFetcher instanceMetadataFetcher,
RingTopologyRefresher ringTopologyRefresher,
+ ExecutorPools executorPools,
SidecarMetrics metrics)
{
this.restoreJobConfig = restoreJobConfig;
@@ -127,11 +134,12 @@ public class RestoreJobDiscoverer implements PeriodicTask
this.restoreRangeDatabaseAccessor = restoreRangeDatabaseAccessor;
this.jobDiscoveryRecencyDays = restoreJobConfig.jobDiscoveryMinimumRecencyDays();
this.restoreJobManagerGroupSingleton = restoreJobManagerGroupProvider;
- this.localTokenRangesProvider = cachedLocalTokenRanges;
this.instanceMetadataFetcher = instanceMetadataFetcher;
this.ringTopologyRefresher = ringTopologyRefresher;
+ this.localTokenRangesProvider = ringTopologyRefresher;
this.metrics = metrics.server().restore();
this.jobIdsByDay = new JobIdsByDay();
+ this.executorPool = executorPools.internal();
}
@Override
@@ -240,6 +248,78 @@ public class RestoreJobDiscoverer implements PeriodicTask
this.periodicTaskExecutor = executor;
}
+ @Override
+ public void onRingTopologyChanged(String keyspace, TokenRangeReplicasResponse oldTopology, TokenRangeReplicasResponse newTopology)
+ {
+ if (oldTopology == null)
+ {
+ LOGGER.debug("Received RingTopologyChanged notification for new topology discovered. " +
+ "It is already handled inline at findSlicesAndSubmit. Exiting early. " +
+ "keyspace={}", keyspace);
+ return;
+ }
+
+ Map<Integer, Set<TokenRange>> localRangesFromOld = RingTopologyRefresher.calculateLocalTokenRanges(instanceMetadataFetcher, oldTopology);
+ Map<Integer, Set<TokenRange>> localRangesFromNew = RingTopologyRefresher.calculateLocalTokenRanges(instanceMetadataFetcher, newTopology);
+ if (Objects.equals(localRangesFromOld, localRangesFromNew))
+ {
+ LOGGER.debug("Local token ranges derived from both topology are the same. No need to update restore ranges.");
+ return;
+ }
+
+ // Populate the lostRanges and the gainedRanges
+ // For each lost range, we want to cancel the RestoreRange that covers it
+ // For each gained range, we want to create the RestoreRange
+ Map<Integer, Set<TokenRange>> lostRanges = new HashMap<>(localRangesFromNew.size());
+ Map<Integer, Set<TokenRange>> gainedRanges = new HashMap<>(localRangesFromNew.size());
+ for (Integer instanceId : Sets.union(localRangesFromOld.keySet(), localRangesFromNew.keySet()))
+ {
+ Set<TokenRange> rangesFromOld = localRangesFromOld.get(instanceId);
+ Set<TokenRange> rangesFromNew = localRangesFromNew.get(instanceId);
+ Preconditions.checkState(rangesFromNew != null || rangesFromOld != null,
+ "Token ranges of instance: " + instanceId + " do not exist in both old and new");
+ if (rangesFromOld == null) // new node
+ {
+ gainedRanges.put(instanceId, rangesFromNew);
+ }
+ else if (rangesFromNew == null) // removed node
+ {
+ lostRanges.put(instanceId, rangesFromOld);
+ }
+ else // both new and old ranges exist and they differs
+ {
+ TokenRange.SymmetricDiffResult symmetricDiffResult = TokenRange.symmetricDiff(rangesFromOld, rangesFromNew);
+ // ranges that are no longer in the new topology are lost
+ lostRanges.put(instanceId, symmetricDiffResult.onlyInLeft);
+ // ranges that are new in the new topology are gained
+ gainedRanges.put(instanceId, symmetricDiffResult.onlyInRight);
+ }
+ }
+
+ Set<UUID> jobIds = ringTopologyRefresher.allRestoreJobsOfKeyspace(keyspace);
+ for (UUID jobId : jobIds)
+ {
+ RestoreJob restoreJob = restoreJobDatabaseAccessor.find(jobId);
+ if (restoreJob == null)
+ {
+ continue;
+ }
+
+ try
+ {
+ // First, discard all the restore ranges that cover the lost ranges
+ lostRanges.forEach((instanceId, ranges) -> discardLostRanges(restoreJob, instanceId, ranges));
+ // Next, submit the RestoreRanges from the newly gained ranges
+ gainedRanges.forEach((instanceId, ranges) -> findSlicesOfCassandraNodeAndSubmit(restoreJob, instanceId, ranges));
+ }
+ catch (Exception e)
+ {
+ // log the warning and continue to process other jobs
+ LOGGER.warn("Unexpected exception when adjusting restore job ranges. jobId={}", jobId, e);
+ }
+ }
+ }
+
private void processOneJob(RestoreJob job, RestoreJobManagerGroup restoreJobManagers, RunContext context)
{
if (jobIdsByDay.shouldLogJob(job))
@@ -249,31 +329,21 @@ public class RestoreJobDiscoverer implements PeriodicTask
switch (job.status)
{
+ case STAGED:
+ // unset the flag, so that it can re-discover the slices when the job status changes to a ready status
+ jobIdsByDay.unsetSlicesDiscovered(job); // no break by design
case CREATED:
case STAGE_READY:
- case STAGED:
case IMPORT_READY:
if (job.hasExpired(context.nowMillis))
{
- context.expiredJobs += 1;
- boolean aborted = abortJob(job);
- if (aborted)
- {
- // finalize the job once aborted; otherwise retry in the next periodic task run
- context.abortedJobs += 1;
- finalizeJob(restoreJobManagers, job);
- }
+ abortExpiredJob(job, restoreJobManagers, context);
break; // do not proceed further if the job has expired
}
// find the oldest non-completed job
context.earliestInDays = Math.max(context.earliestInDays, delta(context.today, job.createdAt));
restoreJobManagers.updateRestoreJob(job);
- if (job.isManagedBySidecar())
- {
- ringTopologyRefresher.register(job);
- // todo: potential exceedingly number of queries
- findSlicesAndSubmit(job);
- }
+ processSidecarManagedJobMaybe(job);
inflightJobsCount += 1;
break;
case FAILED:
@@ -286,12 +356,48 @@ public class RestoreJobDiscoverer implements PeriodicTask
}
}
+ private void abortExpiredJob(RestoreJob job, RestoreJobManagerGroup restoreJobManagers, RunContext context)
+ {
+ context.expiredJobs += 1;
+ boolean aborted = abortJob(job);
+ if (aborted)
+ {
+ // finalize the job once aborted; otherwise retry in the next periodic task run
+ context.abortedJobs += 1;
+ finalizeJob(restoreJobManagers, job);
+ }
+ }
+
+ private void processSidecarManagedJobMaybe(RestoreJob job)
+ {
+ if (!job.isManagedBySidecar())
+ {
+ return;
+ }
+
+ // Only force refresh topology for the first time in each stage
+ // RestoreJobDiscoverer is registered as a RingTopologyListener to receive future topology changed notifications, if any
+ ringTopologyRefresher.register(job, this);
+ if (shouldFindSlicesAndSubmit(job))
+ {
+ findSlicesAndSubmit(job);
+ // Mark the flag. It prevents finding slices (which is expensive) until the flag is unset.
+ jobIdsByDay.markSlicesDiscovered(job);
+ }
+ }
+
+ private boolean shouldFindSlicesAndSubmit(RestoreJob job)
+ {
+ return (job.status == RestoreJobStatus.STAGE_READY || job.status == RestoreJobStatus.IMPORT_READY)
+ && !jobIdsByDay.isSliceDiscovered(job);
+ }
+
private void finalizeJob(RestoreJobManagerGroup restoreJobManagers, RestoreJob job)
{
restoreJobManagers.removeJobInternal(job);
if (job.isManagedBySidecar())
{
- ringTopologyRefresher.unregister(job);
+ ringTopologyRefresher.unregister(job, this);
}
}
@@ -299,12 +405,15 @@ public class RestoreJobDiscoverer implements PeriodicTask
// according to the cluster token ownership
private void findSlicesAndSubmit(RestoreJob restoreJob)
{
- localTokenRangesProvider.localTokenRanges(restoreJob.keyspaceName)
- .forEach((key, ranges) -> {
- int instanceId = key;
- InstanceMetadata instance = instanceMetadataFetcher.instance(instanceId);
- ranges.forEach(range -> findSlicesOfRangeAndSubmit(instance, restoreJob, range));
- });
+ localTokenRangesProvider.localTokenRanges(restoreJob.keyspaceName, true)
+ .forEach((instanceId, ranges) -> findSlicesOfCassandraNodeAndSubmit(restoreJob, instanceId, ranges));
+ }
+
+ // find all slices according to the Cassandra node denoted by the instanceId and the token ranges
+ private void findSlicesOfCassandraNodeAndSubmit(RestoreJob restoreJob, int instanceId, Set<TokenRange> ranges)
+ {
+ InstanceMetadata instance = instanceMetadataFetcher.instance(instanceId);
+ ranges.forEach(range -> findSlicesOfRangeAndSubmit(instance, restoreJob, range));
}
// try to submit the slice.
@@ -316,33 +425,50 @@ public class RestoreJobDiscoverer implements PeriodicTask
restoreSliceDatabaseAccessor
.selectByJobByBucketByTokenRange(restoreJob, bucketId, range)
.forEach(slice -> {
- // set the owner instance, which is not read from database
- TokenRangeReplicasResponse topology = ringTopologyRefresher.cachedReplicaByTokenRange(restoreJob);
- List<RestoreRange> splits = slice.splitMaybe(topology)
- .stream()
- .map(s -> {
- String uploadId = RestoreJobUtil.generateUniqueUploadId(s.jobId(), s.sliceId());
- return RestoreRange.builderFromSlice(s)
- .ownerInstance(instance)
- .stageDirectory(Paths.get(instance.stagingDir()), uploadId)
- .build();
- })
- .collect(Collectors.toList());
- RestoreJobManagerGroup managerGroup = restoreJobManagerGroupSingleton.get();
- for (RestoreRange split : splits)
+ // Check if the slice needs to be trimmed/split
+ RestoreSlice trimmed = slice.trimMaybe(range);
+ String uploadId = RestoreJobUtil.generateUniqueUploadId(trimmed.jobId(), trimmed.sliceId());
+ RestoreRange restoreRange = RestoreRange.builderFromSlice(trimmed)
+ // set the owner instance, which is not read from database
+ .ownerInstance(instance)
+ .stageDirectory(Paths.get(instance.stagingDir()), uploadId)
+ .build();
+ RestoreJobProgressTracker.Status status = submit(instance, restoreJob, restoreRange);
+ if (status == RestoreJobProgressTracker.Status.CREATED)
{
- RestoreJobProgressTracker.Status status = submit(instance, managerGroup, restoreJob, split);
- if (status == RestoreJobProgressTracker.Status.CREATED)
- {
- restoreRangeDatabaseAccessor.create(split);
- }
+ restoreRangeDatabaseAccessor.create(restoreRange);
}
});
}
- private RestoreJobProgressTracker.Status submit(InstanceMetadata instance, RestoreJobManagerGroup managerGroup,
- RestoreJob job, RestoreRange range)
+ private void discardLostRanges(RestoreJob restoreJob, int instanceId, Set<TokenRange> otherRanges)
{
+ InstanceMetadata instance = instanceMetadataFetcher.instance(instanceId);
+ RestoreJobManagerGroup managerGroup = restoreJobManagerGroupSingleton.get();
+ Set<RestoreRange> overlappingRanges = managerGroup.discardOverlappingRanges(instance, restoreJob, otherRanges);
+ calculateRemainingRangesAndResubmit(restoreJob, instanceId, otherRanges, overlappingRanges);
+ }
+
+ // There could be still ranges remaining after subtracting the overlapping parts.
+ // The method calculates the ranges that still remain to the Cassandra node (identified by instanceId), and
+ // re-submit the RestoreRange for the remaining ranges.
+ private void calculateRemainingRangesAndResubmit(RestoreJob restoreJob,
+ int instanceId,
+ Set<TokenRange> otherRanges,
+ Set<RestoreRange> overlappingRanges)
+ {
+ Set<TokenRange> existingRanges = overlappingRanges.stream()
+ .map(RestoreRange::tokenRange)
+ .collect(Collectors.toSet());
+ TokenRange.SymmetricDiffResult symmetricDiffResult = TokenRange.symmetricDiff(existingRanges, otherRanges);
+ Set<TokenRange> remainedRanges = symmetricDiffResult.onlyInLeft;
+ findSlicesOfCassandraNodeAndSubmit(restoreJob, instanceId, remainedRanges);
+ }
+
+ private RestoreJobProgressTracker.Status submit(InstanceMetadata instance, RestoreJob job, RestoreRange range)
+ {
+ RestoreJobManagerGroup managerGroup = restoreJobManagerGroupSingleton.get();
+
try
{
return managerGroup.trySubmit(instance, range, job);
@@ -381,6 +507,8 @@ public class RestoreJobDiscoverer implements PeriodicTask
static class JobIdsByDay
{
private final Map<Integer, Map<UUID, RestoreJobStatus>> jobsByDay = new HashMap<>();
+ // tracks the jobIds that have their slices already discovered
+ private final Map<Integer, Set<UUID>> sliceDiscoveredJobsByDay = new HashMap<>();
private final Set<Integer> discoveredDays = new HashSet<>(); // contains the days of the jobs seen from the current round of discovery
/**
@@ -393,13 +521,46 @@ public class RestoreJobDiscoverer implements PeriodicTask
*/
boolean shouldLogJob(RestoreJob job)
{
- int day = job.createdAt.getDaysSinceEpoch();
- discoveredDays.add(day);
+ int day = populateDiscoveredDay(job);
Map<UUID, RestoreJobStatus> jobs = jobsByDay.computeIfAbsent(day, key -> new HashMap<>());
RestoreJobStatus oldStatus = jobs.put(job.jobId, job.status);
return oldStatus == null || job.status == RestoreJobStatus.CREATED || oldStatus != job.status;
}
+ void markSlicesDiscovered(RestoreJob job)
+ {
+ int day = populateDiscoveredDay(job);
+ sliceDiscoveredJobsByDay.compute(day, (key, value) -> {
+ if (value == null)
+ {
+ value = new HashSet<>();
+ }
+ value.add(job.jobId);
+ return value;
+ });
+ }
+
+ boolean isSliceDiscovered(RestoreJob job)
+ {
+ int day = populateDiscoveredDay(job);
+ return sliceDiscoveredJobsByDay.getOrDefault(day, Collections.emptySet())
+ .contains(job.jobId);
+ }
+
+ void unsetSlicesDiscovered(RestoreJob job)
+ {
+ int day = populateDiscoveredDay(job);
+ sliceDiscoveredJobsByDay.compute(day, (key, value) -> {
+ if (value == null)
+ {
+ return null;
+ }
+
+ value.remove(job.jobId);
+ return value;
+ });
+ }
+
void cleanupMaybe()
{
// remove all the jobIds of the days that are not discovered
@@ -407,6 +568,13 @@ public class RestoreJobDiscoverer implements PeriodicTask
discoveredDays.clear();
}
+ private int populateDiscoveredDay(RestoreJob job)
+ {
+ int day = job.createdAt.getDaysSinceEpoch();
+ discoveredDays.add(day);
+ return day;
+ }
+
@VisibleForTesting
Map<Integer, Map<UUID, RestoreJobStatus>> jobsByDay()
{
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobManager.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobManager.java
index 8b39631d..ec81a492 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobManager.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobManager.java
@@ -25,6 +25,7 @@ import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Comparator;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
@@ -37,6 +38,7 @@ import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import io.vertx.core.Future;
import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.common.server.utils.ThrowableUtils;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
import org.apache.cassandra.sidecar.config.RestoreJobConfiguration;
@@ -112,11 +114,22 @@ public class RestoreJobManager
*/
void updateRestoreJob(RestoreJob restoreJob)
{
- RestoreJobProgressTracker tracker = jobs.computeIfAbsent(restoreJob.jobId,
- id -> new RestoreJobProgressTracker(restoreJob, processor, instanceMetadata));
+ RestoreJobProgressTracker tracker = progressTracker(restoreJob);
tracker.updateRestoreJob(restoreJob);
}
+ /**
+ * Discard all the {@link RestoreRange} that overlap with {@param otherRanges} in the {@link RestoreJob}
+ * @param restoreJob restore job to find out the restore ranges
+ * @param otherRanges token ranges to find the overlapping {@link RestoreRange} and discard
+ * @return set of overlapping {@link RestoreRange}
+ */
+ Set<RestoreRange> discardOverlappingRanges(RestoreJob restoreJob, Set<TokenRange> otherRanges)
+ {
+ RestoreJobProgressTracker tracker = progressTracker(restoreJob);
+ return tracker.discardOverlappingRanges(otherRanges);
+ }
+
/**
* Remove the tracker of the job when it is completed and delete its data on disk. The method runs async and it for internal use only.
* It should only be called by the background task, when it discovers the job is
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobManagerGroup.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobManagerGroup.java
index da36f5da..52a6ecd4 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobManagerGroup.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobManagerGroup.java
@@ -19,13 +19,18 @@
package org.apache.cassandra.sidecar.restore;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import com.google.inject.Inject;
import com.google.inject.Singleton;
import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
import org.apache.cassandra.sidecar.config.RestoreJobConfiguration;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
@@ -40,6 +45,8 @@ import org.apache.cassandra.sidecar.tasks.PeriodicTaskExecutor;
@Singleton
public class RestoreJobManagerGroup
{
+ private static final Logger LOGGER = LoggerFactory.getLogger(RestoreJobManagerGroup.class);
+
private final RestoreJobConfiguration restoreJobConfig;
// instance id --> RestoreJobManager
private final Map<Integer, RestoreJobManager> managerGroup = new ConcurrentHashMap<>();
@@ -114,6 +121,29 @@ public class RestoreJobManagerGroup
managerGroup.values().forEach(manager -> manager.updateRestoreJob(restoreJob));
}
+ /**
+ * Discard the ranges that overlap with the given {@param otherRanges}
+ * @param instanceMetadata cassandra instance to discard the restore range from
+ * @param restoreJob restore job instance
+ * @param otherRanges set of {@link TokenRange} to find the overlapping {@link RestoreRange} and discard
+ * @return set of overlapping {@link RestoreRange}
+ */
+ Set<RestoreRange> discardOverlappingRanges(InstanceMetadata instanceMetadata, RestoreJob restoreJob, Set<TokenRange> otherRanges)
+ {
+ if (restoreJob.status.isFinal())
+ {
+ throw new IllegalStateException("Cannot remove ranges from a restore job in final status");
+ }
+ RestoreJobManager manager = managerGroup.get(instanceMetadata.id());
+ if (manager == null)
+ {
+ LOGGER.debug("No RestoreJobManager found for Cassandra instance. No ranges to discard. instanceId={}",
+ instanceMetadata.id());
+ return Set.of();
+ }
+ return manager.discardOverlappingRanges(restoreJob, otherRanges);
+ }
+
/**
* @return the {@link RestoreJobManager} of the instance
*/
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobProgress.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobProgress.java
index 539fb772..94473eec 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobProgress.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobProgress.java
@@ -18,6 +18,7 @@ package org.apache.cassandra.sidecar.restore;
import java.util.ArrayList;
import java.util.List;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
import org.apache.cassandra.sidecar.common.DataObjectBuilder;
@@ -28,6 +29,7 @@ import org.apache.cassandra.sidecar.common.response.data.RestoreRangeJson;
import org.apache.cassandra.sidecar.db.RestoreJob;
import org.apache.cassandra.sidecar.db.RestoreRange;
import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.VisibleForTesting;
/**
* Represents the progress of a {@link RestoreJob}
@@ -106,6 +108,22 @@ public class RestoreJobProgress
return message + " Current job status: " + jobStatus;
}
+ @VisibleForTesting
+ List<RestoreRange> allRanges()
+ {
+ List<RestoreRange> result = new ArrayList<>();
+ Consumer<List<RestoreRange>> addAllIgnoreNull = list -> {
+ if (list != null)
+ {
+ result.addAll(list);
+ }
+ };
+ addAllIgnoreNull.accept(failedRanges);
+ addAllIgnoreNull.accept(pendingRanges);
+ addAllIgnoreNull.accept(succeededRanges);
+ return result;
+ }
+
static class Builder implements DataObjectBuilder<Builder, RestoreJobProgress>
{
private final RestoreJob restoreJob;
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobProgressTracker.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobProgressTracker.java
index 8e7057e7..ee38c23f 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobProgressTracker.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobProgressTracker.java
@@ -18,21 +18,28 @@
package org.apache.cassandra.sidecar.restore;
+import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
+import com.google.common.collect.RangeSet;
+import com.google.common.collect.TreeRangeSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.Token;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.db.RestoreJob;
import org.apache.cassandra.sidecar.db.RestoreRange;
import org.apache.cassandra.sidecar.db.RestoreSlice;
import org.apache.cassandra.sidecar.exceptions.RestoreJobFatalException;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.VisibleForTesting;
/**
* In-memory only tracker that tracks the progress of the slices in a restore job.
@@ -67,13 +74,13 @@ public class RestoreJobProgressTracker
if (failureRef.get() != null)
throw failureRef.get();
- Status status = ranges.putIfAbsent(range, Status.PENDING);
+ RestoreRange rangeWithTracker = range.unbuild()
+ .restoreJobProgressTracker(this)
+ .build();
+ Status status = ranges.putIfAbsent(rangeWithTracker, Status.PENDING);
if (status == null)
{
- RestoreRange rangeWithTracker = range.unbuild()
- .restoreJobProgressTracker(this)
- .build();
processor.submit(rangeWithTracker);
return Status.CREATED;
}
@@ -81,14 +88,27 @@ public class RestoreJobProgressTracker
return status;
}
- void removeRange(RestoreRange range)
+ /**
+ * Discard all the {@link RestoreRange} that overlap with the {@param otherRanges}
+ * @param otherRanges token ranges to find the overlapping {@link RestoreRange} and discard
+ * @return set of overlapping {@link RestoreRange}
+ */
+ @SuppressWarnings("UnstableApiUsage")
+ Set<RestoreRange> discardOverlappingRanges(Set<TokenRange> otherRanges)
{
- Status status = ranges.remove(range);
-
- if (status != null)
- {
- processor.remove(range);
- }
+ RangeSet<Token> rangeSet = TreeRangeSet.create();
+ otherRanges.forEach(r -> rangeSet.add(r.range));
+ Set<RestoreRange> overlapping = new HashSet<>();
+ ranges.keySet().removeIf(restoreRange -> {
+ if (rangeSet.intersects(restoreRange.tokenRange().range))
+ {
+ overlapping.add(restoreRange);
+ processor.discardAndRemove(restoreRange);
+ return true;
+ }
+ return false;
+ });
+ return overlapping;
}
void updateRestoreJob(@NotNull RestoreJob restoreJob)
@@ -170,8 +190,14 @@ public class RestoreJobProgressTracker
}
}
+ @VisibleForTesting // do not call in production code
+ Map<RestoreRange, Status> rangesForTesting()
+ {
+ return ranges;
+ }
+
/**
- * Enum holds possible statues of {@link RestoreSlice}
+ * Task status of {@link RestoreRangeTask}
*/
public enum Status
{
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreProcessor.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreProcessor.java
index 429ecc27..0a72ad9a 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreProcessor.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreProcessor.java
@@ -125,12 +125,15 @@ public class RestoreProcessor implements PeriodicTask
*
* @param range restore range to be removed
*/
- void remove(RestoreRange range)
+ void discardAndRemove(RestoreRange range)
{
if (isClosed)
return;
workQueue.remove(range);
+ range.discard();
+ // update the range async
+ pool.runBlocking(() -> rangeDatabaseAccessor.updateStatus(range));
}
@Override
@@ -283,7 +286,12 @@ public class RestoreProcessor implements PeriodicTask
private Function<Throwable, Future<Void>> taskFailureHandler(RestoreRange range)
{
return cause -> {
- if (cause instanceof RestoreJobException && ((RestoreJobException) cause).retryable())
+ if (range.isDiscarded())
+ {
+ // for discarded ranges, we simply do not re-queue them.
+ LOGGER.debug("RestoreRange is discarded. sliceKey={}", range.sliceKey());
+ }
+ else if (cause instanceof RestoreJobException && ((RestoreJobException) cause).retryable())
{
LOGGER.warn("Slice failed with recoverable failure. sliceKey={}", range.sliceKey(), cause);
// re-enqueue the retryable failed slice
@@ -312,7 +320,7 @@ public class RestoreProcessor implements PeriodicTask
}
@VisibleForTesting
- int activeSlices()
+ int activeRanges()
{
return workQueue.activeRangesCount();
}
@@ -324,7 +332,7 @@ public class RestoreProcessor implements PeriodicTask
}
@VisibleForTesting
- int pendingStartSlices()
+ int pendingStartRanges()
{
return workQueue.size();
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreRangeTask.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreRangeTask.java
index 469cd403..203d011a 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreRangeTask.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreRangeTask.java
@@ -492,7 +492,7 @@ public class RestoreRangeTask implements RestoreRangeHandler
break;
}
- hasOverlap = owningRange.overlaps(sstableRange);
+ hasOverlap = owningRange.intersects(sstableRange);
if (hasOverlap)
{
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RingTopologyChangeListener.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RingTopologyChangeListener.java
new file mode 100644
index 00000000..202eb9b7
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RingTopologyChangeListener.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.restore;
+
+import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * To receive notifications on ring topology changed
+ */
+public interface RingTopologyChangeListener
+{
+ /**
+ * Notification on ring topology changed
+ * @param keyspace keyspace (which defines replication factor) to derive the topology from
+ * @param oldTopology old topology. The value is nullable. When the ring topology of the keyspace is just learned, the value is null.
+ * @param newTopology new topology. The value is always non-null
+ */
+ void onRingTopologyChanged(String keyspace,
+ @Nullable TokenRangeReplicasResponse oldTopology,
+ @NotNull TokenRangeReplicasResponse newTopology);
+}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RingTopologyRefresher.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RingTopologyRefresher.java
index d1a702c9..6a845b36 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RingTopologyRefresher.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RingTopologyRefresher.java
@@ -18,13 +18,19 @@
package org.apache.cassandra.sidecar.restore;
+import java.net.UnknownHostException;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiFunction;
import java.util.function.Function;
+import java.util.stream.Collectors;
+import javax.annotation.concurrent.ThreadSafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,39 +40,52 @@ import com.google.inject.Singleton;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.cluster.locator.LocalTokenRangesProvider;
import org.apache.cassandra.sidecar.common.response.NodeSettings;
import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.Token;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.common.server.data.Name;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
+import org.apache.cassandra.sidecar.common.server.utils.StringUtils;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
import org.apache.cassandra.sidecar.config.RestoreJobConfiguration;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
import org.apache.cassandra.sidecar.db.RestoreJob;
import org.apache.cassandra.sidecar.tasks.PeriodicTask;
import org.apache.cassandra.sidecar.tasks.ScheduleDecision;
import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.VisibleForTesting;
/**
* Refreshes the Cassandra ring topology fetched via JMX periodically
+ * // TODO: this class can be generalized to serve other Sidecar components that need to be aware of topology and topology change
*/
@Singleton
-public class RingTopologyRefresher implements PeriodicTask
+public class RingTopologyRefresher implements PeriodicTask, LocalTokenRangesProvider
{
private static final Logger LOGGER = LoggerFactory.getLogger(RingTopologyRefresher.class);
private final InstanceMetadataFetcher metadataFetcher;
private final ReplicaByTokenRangePerKeyspace replicaByTokenRangePerKeyspace;
private final RestoreJobConfiguration restoreJobConfiguration;
+ private final TaskExecutorPool executorPool;
+ private final Map<String, Set<RingTopologyChangeListener>> listenersByKeyspace = new ConcurrentHashMap<>();
@Inject
public RingTopologyRefresher(InstanceMetadataFetcher metadataFetcher,
- SidecarConfiguration config)
+ SidecarConfiguration config,
+ ExecutorPools executorPools)
{
this.metadataFetcher = metadataFetcher;
this.restoreJobConfiguration = config.restoreJobConfiguration();
- this.replicaByTokenRangePerKeyspace = new ReplicaByTokenRangePerKeyspace();
+ this.replicaByTokenRangePerKeyspace = new ReplicaByTokenRangePerKeyspace(this::dispatchRingTopologyChangeAsync);
+ this.executorPool = executorPools.internal();
}
@Override
@@ -81,21 +100,51 @@ public class RingTopologyRefresher implements PeriodicTask
return replicaByTokenRangePerKeyspace.isEmpty() ? ScheduleDecision.SKIP : ScheduleDecision.EXECUTE;
}
+ /**
+ * Execute the periodic task to refresh the topology layout for all registered keyspaces
+ *
+ * <p>It is synchronized as there is potential contention from {@link #localTokenRanges(String)}
+ * @param promise a promise when the execution completes
+ */
@Override
- public void execute(Promise<Void> promise)
+ public synchronized void execute(Promise<Void> promise)
{
prepareAndFetch(this::loadAll);
promise.tryComplete();
}
- public void register(RestoreJob restoreJob)
+ public Set<UUID> allRestoreJobsOfKeyspace(String keyspace)
+ {
+ return replicaByTokenRangePerKeyspace.jobsByKeyspace.get(keyspace);
+ }
+
+ public void register(RestoreJob restoreJob, RingTopologyChangeListener listener)
{
replicaByTokenRangePerKeyspace.register(restoreJob);
+ addRingTopologyChangeListener(restoreJob.keyspaceName, listener);
+ }
+
+ public void unregister(RestoreJob restoreJob, RingTopologyChangeListener listener)
+ {
+ boolean allRemoved = replicaByTokenRangePerKeyspace.unregister(restoreJob);
+ if (allRemoved)
+ {
+ removeRingTopologyChangeListener(restoreJob.keyspaceName, listener);
+ }
}
- public void unregister(RestoreJob restoreJob)
+ public void addRingTopologyChangeListener(String keyspace, RingTopologyChangeListener listener)
{
- replicaByTokenRangePerKeyspace.unregister(restoreJob);
+ listenersByKeyspace.computeIfAbsent(keyspace, key -> ConcurrentHashMap.newKeySet())
+ .add(listener);
+ }
+
+ public void removeRingTopologyChangeListener(String keyspace, RingTopologyChangeListener listener)
+ {
+ listenersByKeyspace.computeIfPresent(keyspace, (k, v) -> {
+ v.remove(listener);
+ return v.isEmpty() ? null : v;
+ });
}
@Nullable
@@ -104,6 +153,79 @@ public class RingTopologyRefresher implements PeriodicTask
return replicaByTokenRangePerKeyspace.forRestoreJob(restoreJob);
}
+ /**
+ * Fetch the latest topology view
+ * <p>It is synchronized when force refreshing as there is potential contention from {@link #execute(Promise)}
+ *
+ * @param keyspace keyspace to determine replication
+ * @param forceRefresh whether refresh the topology view forcibly or not
+ * @return token ranges of the local Cassandra instances or an empty map of nothing is found
+ */
+ @Override
+ public Map<Integer, Set<TokenRange>> localTokenRanges(String keyspace, boolean forceRefresh)
+ {
+ TokenRangeReplicasResponse topology;
+ if (forceRefresh) // fetch the latest topology and load into cache
+ {
+ synchronized (this)
+ {
+ topology = prepareAndFetch((storageOperations, nodeSettings) -> {
+ String partitioner = nodeSettings.partitioner();
+ return replicaByTokenRangePerKeyspace.loadOne(keyspace, k -> storageOperations.tokenRangeReplicas(new Name(keyspace), partitioner));
+ });
+ }
+ }
+ else // get the cached value
+ {
+ topology = replicaByTokenRangePerKeyspace.topologyOfKeyspace(keyspace);
+ }
+
+ return calculateLocalTokenRanges(metadataFetcher, topology);
+ }
+
+ // todo: refactor to a utility class _when_ refactoring TokenRangeReplicasResponse data structure (separate out server and http data representations)
+ @NotNull
+ public static Map<Integer, Set<TokenRange>> calculateLocalTokenRanges(InstanceMetadataFetcher metadataFetcher, TokenRangeReplicasResponse topology)
+ {
+ if (topology == null)
+ {
+ return Map.of();
+ }
+
+ // todo: this assumes one C* node per IP address
+ Map<String, Integer> allNodes = topology.replicaMetadata().values().stream()
+ .collect(Collectors.toMap(TokenRangeReplicasResponse.ReplicaMetadata::address,
+ TokenRangeReplicasResponse.ReplicaMetadata::port));
+
+ List<InstanceMetadata> localNodes = metadataFetcher.allLocalInstances();
+ Map<String, InstanceMetadata> localEndpointsToMetadata = new HashMap<>(localNodes.size());
+ for (InstanceMetadata instanceMetadata : localNodes)
+ {
+ populateEndpointToMetadata(instanceMetadata, allNodes, localEndpointsToMetadata);
+ }
+
+ Map<Integer, Set<TokenRange>> localTokenRanges = new HashMap<>(localEndpointsToMetadata.size());
+ for (TokenRangeReplicasResponse.ReplicaInfo ri : topology.writeReplicas())
+ {
+ TokenRange range = new TokenRange(Token.from(ri.start()), Token.from(ri.end()));
+ for (List<String> instanceOfDc : ri.replicasByDatacenter().values())
+ {
+ for (String instanceEndpoint : instanceOfDc)
+ {
+ InstanceMetadata instanceMetadata = localEndpointsToMetadata.get(instanceEndpoint);
+ if (instanceMetadata == null)
+ {
+ // skip the non-local nodes
+ continue;
+ }
+ localTokenRanges.computeIfAbsent(instanceMetadata.id(), k -> new HashSet<>())
+ .add(range);
+ }
+ }
+ }
+ return localTokenRanges;
+ }
+
/**
* Retrieve {@link TokenRangeReplicasResponse} that matters to the input restoreJob asynchronously
* If {@link RingTopologyRefresher} has not retrieved anything yet, the returned future should reflect the initial
@@ -139,18 +261,65 @@ public class RingTopologyRefresher implements PeriodicTask
});
}
+ private void dispatchRingTopologyChangeAsync(String keyspace, TokenRangeReplicasResponse oldTopology, TokenRangeReplicasResponse newTopology)
+ {
+ // Dispatch onRingTopologyChanged in another thread as it might block
+ // This method cannot block, since it is invoked by org.apache.cassandra.sidecar.restore.RingTopologyRefresher.ReplicaByTokenRangePerKeyspace.loadOne
+ // in the `compute` lambda.
+ listenersByKeyspace.getOrDefault(keyspace, Collections.emptySet()).forEach(listener -> {
+ executorPool.runBlocking(() -> listener.onRingTopologyChanged(keyspace, oldTopology, newTopology));
+ });
+ }
+
+ private static void populateEndpointToMetadata(InstanceMetadata instanceMetadata,
+ Map<String, Integer> allNodes,
+ Map<String, InstanceMetadata> localEndpointsToMetadata)
+ {
+ String ip = ipOf(instanceMetadata);
+ Integer port = allNodes.get(ip);
+ if (ip == null || port == null)
+ {
+ // the configured local instance is not in the C* ring yet (not even joining). Exit early for such node
+ return;
+ }
+ String endpointWithPort = StringUtils.cassandraFormattedHostAndPort(ip, port);
+ localEndpointsToMetadata.put(endpointWithPort, instanceMetadata);
+ }
+
+ private static String ipOf(InstanceMetadata instanceMetadata)
+ {
+ String ipAddress = instanceMetadata.ipAddress();
+ if (ipAddress == null)
+ {
+ try
+ {
+ return instanceMetadata.refreshIpAddress();
+ }
+ catch (UnknownHostException uhe)
+ {
+ LOGGER.debug("Failed to resolve IP address. host={}", instanceMetadata.host());
+ }
+ }
+ return ipAddress;
+ }
+
/**
* Core data class of the {@link RingTopologyRefresher}.
* It groups the ReplicaByTokenRange by keyspace and provides cache capability
- * <p>
- * Not thread-safe; only accessed within the single-threaded execution of RingTopologyRefresher
*/
+ @ThreadSafe
static class ReplicaByTokenRangePerKeyspace
{
- private final Set<UUID> allJobs = new HashSet<>();
- private final Map<String, Set<UUID>> jobsByKeyspace = new HashMap<>();
- private final Map<String, TokenRangeReplicasResponse> mapping = new HashMap<>();
- private final Map<String, Promise<TokenRangeReplicasResponse>> promises = new HashMap<>();
+ private final Set<UUID> allJobs = ConcurrentHashMap.newKeySet();
+ private final Map<String, Set<UUID>> jobsByKeyspace = new ConcurrentHashMap<>();
+ private final Map<String, TokenRangeReplicasResponse> mapping = new ConcurrentHashMap<>();
+ private final Map<String, Promise<TokenRangeReplicasResponse>> promises = new ConcurrentHashMap<>();
+ private final RingTopologyChangeListener asyncDispatcher;
+
+ ReplicaByTokenRangePerKeyspace(@NotNull RingTopologyChangeListener asyncDispatcher)
+ {
+ this.asyncDispatcher = asyncDispatcher;
+ }
boolean isEmpty()
{
@@ -160,19 +329,26 @@ public class RingTopologyRefresher implements PeriodicTask
// returns keyspace name
String register(RestoreJob restoreJob)
{
- allJobs.add(restoreJob.jobId);
- jobsByKeyspace.compute(restoreJob.keyspaceName, (ks, jobs) -> {
- Set<UUID> jobsSet = jobs == null ? new HashSet<>() : jobs;
- jobsSet.add(restoreJob.jobId);
- return jobsSet;
- });
+ // the job is not yet registered
+ if (allJobs.add(restoreJob.jobId))
+ {
+ jobsByKeyspace.computeIfAbsent(restoreJob.keyspaceName, ks -> ConcurrentHashMap.newKeySet())
+ .add(restoreJob.jobId);
+ }
+
return restoreJob.keyspaceName;
}
- void unregister(RestoreJob restoreJob)
+ /**
+ * Unregister a restore job
+ *
+ * @param restoreJob restore job to unregister
+ * @return true when all jobs of the same keyspace are removed; false otherwise
+ */
+ boolean unregister(RestoreJob restoreJob)
{
boolean containsJob = allJobs.remove(restoreJob.jobId);
- if (containsJob)
+ if (containsJob) // enter this block only once per jobId
{
Set<UUID> jobIdsByKeyspace = jobsByKeyspace.get(restoreJob.keyspaceName);
if (jobIdsByKeyspace == null
@@ -180,15 +356,17 @@ public class RingTopologyRefresher implements PeriodicTask
|| !jobIdsByKeyspace.remove(restoreJob.jobId))
{
LOGGER.warn("Unable to find the restore job id to unregister. jobId={}", restoreJob.jobId);
- return;
+ return true;
}
if (!jobIdsByKeyspace.isEmpty())
{
// do not remove mapping if there are still jobs of the keyspace
- return;
+ return false;
}
+ // Multiple threads could reach here and run the following code to remove entries from maps
+ // The removal is idempotent. It is fine to run the code multiple times.
LOGGER.info("All jobs of the keyspace are unregistered. keyspace={}", restoreJob.keyspaceName);
jobsByKeyspace.remove(restoreJob.keyspaceName);
mapping.remove(restoreJob.keyspaceName);
@@ -200,6 +378,7 @@ public class RingTopologyRefresher implements PeriodicTask
"jobId=" + restoreJob.jobId + " keyspace=" + restoreJob.keyspaceName);
}
}
+ return true;
}
// returns the future of TokenRangeReplicasResponse for the restore job
@@ -223,16 +402,18 @@ public class RingTopologyRefresher implements PeriodicTask
{
// loop through the unique names
Set<String> distinctKeyspaces = jobsByKeyspace.keySet();
- distinctKeyspaces.forEach(keyspace -> loadEach(keyspace, loader));
+ distinctKeyspaces.forEach(keyspace -> loadOne(keyspace, loader));
}
- // suppress any Exception when loading topology of each keyspace and continue
- void loadEach(String keyspace, Function<String, TokenRangeReplicasResponse> loader)
+ // suppress any Exception when loading topology of each keyspace and continue; returns null if fails to load
+ TokenRangeReplicasResponse loadOne(String keyspace, Function<String, TokenRangeReplicasResponse> loader)
{
try
{
TokenRangeReplicasResponse topology = loader.apply(keyspace);
+ RingTopologyChangeContext context = new RingTopologyChangeContext(keyspace, topology);
mapping.compute(keyspace, (key, existing) -> {
+ context.existing = existing;
if (existing == null)
{
// fulfill promise after retrieving the initial topology
@@ -241,6 +422,7 @@ public class RingTopologyRefresher implements PeriodicTask
promise.tryComplete(topology);
return promise;
});
+ context.shouldDispatch = true;
return topology;
}
else if (existing.writeReplicas().equals(topology.writeReplicas()))
@@ -251,19 +433,32 @@ public class RingTopologyRefresher implements PeriodicTask
else
{
LOGGER.info("Ring topology of keyspace is changed. keyspace={}", keyspace);
+ context.shouldDispatch = true;
return topology;
}
});
+ if (context.shouldDispatch)
+ {
+ asyncDispatcher.onRingTopologyChanged(context.keyspace, context.existing, context.current);
+ }
+ return topology;
}
- catch (Exception ex)
+ catch (Throwable cause)
{
- LOGGER.warn("Failure during load topology for keyspace. keyspace={}", keyspace, ex);
+ LOGGER.warn("Failure during load topology for keyspace. keyspace={}", keyspace, cause);
promises.computeIfPresent(keyspace, (k, promise) -> {
- promise.tryFail(new IllegalStateException("Failed to load topology for keyspace: " + keyspace, ex));
+ promise.tryFail(new IllegalStateException("Failed to load topology for keyspace: " + keyspace, cause));
// return null to remove the promise
return null;
});
}
+ return null;
+ }
+
+ @Nullable
+ TokenRangeReplicasResponse topologyOfKeyspace(String keyspace)
+ {
+ return mapping.get(keyspace);
}
@VisibleForTesting
@@ -289,5 +484,19 @@ public class RingTopologyRefresher implements PeriodicTask
{
return promises;
}
+
+ private static class RingTopologyChangeContext
+ {
+ final String keyspace;
+ final TokenRangeReplicasResponse current;
+ TokenRangeReplicasResponse existing;
+ boolean shouldDispatch = false;
+
+ RingTopologyChangeContext(String keyspace, TokenRangeReplicasResponse current)
+ {
+ this.keyspace = keyspace;
+ this.current = current;
+ }
+ }
}
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/routes/restore/RestoreJobProgressHandler.java b/server/src/main/java/org/apache/cassandra/sidecar/routes/restore/RestoreJobProgressHandler.java
index 856e7fb2..2080c4f7 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/routes/restore/RestoreJobProgressHandler.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/routes/restore/RestoreJobProgressHandler.java
@@ -32,7 +32,7 @@ import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
import org.apache.cassandra.sidecar.common.data.RestoreJobProgressFetchPolicy;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
import org.apache.cassandra.sidecar.db.RestoreJob;
-import org.apache.cassandra.sidecar.restore.RestoreJobConsistencyLevelChecker;
+import org.apache.cassandra.sidecar.restore.RestoreJobConsistencyChecker;
import org.apache.cassandra.sidecar.restore.RestoreJobProgress;
import org.apache.cassandra.sidecar.routes.AbstractHandler;
import org.apache.cassandra.sidecar.routes.AccessProtected;
@@ -51,7 +51,7 @@ import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpExceptio
@Singleton
public class RestoreJobProgressHandler extends AbstractHandler<RestoreJobProgressFetchPolicy> implements AccessProtected
{
- private final RestoreJobConsistencyLevelChecker consistencyLevelChecker;
+ private final RestoreJobConsistencyChecker consistencyLevelChecker;
/**
* Constructs a handler with the provided {@code metadataFetcher}
@@ -64,7 +64,7 @@ public class RestoreJobProgressHandler extends AbstractHandler<RestoreJobProgres
public RestoreJobProgressHandler(InstanceMetadataFetcher metadataFetcher,
ExecutorPools executorPools,
CassandraInputValidator validator,
- RestoreJobConsistencyLevelChecker consistencyLevelChecker)
+ RestoreJobConsistencyChecker consistencyLevelChecker)
{
super(metadataFetcher, executorPools, validator);
this.consistencyLevelChecker = consistencyLevelChecker;
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/db/RestoreRangeDatabaseAccessorIntTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/db/RestoreRangeDatabaseAccessorIntTest.java
index 1d3b651e..3872530e 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/db/RestoreRangeDatabaseAccessorIntTest.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/db/RestoreRangeDatabaseAccessorIntTest.java
@@ -57,7 +57,7 @@ class RestoreRangeDatabaseAccessorIntTest extends IntegrationTestBase
Map<String, RestoreRangeStatus> statusByReplica = new HashMap<>(fetchedRange.statusByReplica());
assertThat(statusByReplica)
.hasSize(1)
- .containsEntry("replica1", RestoreRangeStatus.CREATED);
+ .containsEntry("127.0.0.1:12345", RestoreRangeStatus.CREATED);
// find no range from non-existing bucket
assertThat(accessor.findAll(jobId, (short) 1))
@@ -65,7 +65,7 @@ class RestoreRangeDatabaseAccessorIntTest extends IntegrationTestBase
.isEmpty();
// update status
- statusByReplica.put("replica1", RestoreRangeStatus.STAGED);
+ statusByReplica.put("127.0.0.1:12345", RestoreRangeStatus.STAGED);
accessor.updateStatus(range.unbuild().replicaStatus(statusByReplica).build());
// read the updated range back
@@ -76,7 +76,7 @@ class RestoreRangeDatabaseAccessorIntTest extends IntegrationTestBase
statusByReplica = new HashMap<>(fetchedRange.statusByReplica());
assertThat(statusByReplica)
.hasSize(1)
- .containsEntry("replica1", RestoreRangeStatus.STAGED);
+ .containsEntry("127.0.0.1:12345", RestoreRangeStatus.STAGED);
// update status with new replica status
statusByReplica.put("replica2", RestoreRangeStatus.CREATED);
@@ -97,7 +97,7 @@ class RestoreRangeDatabaseAccessorIntTest extends IntegrationTestBase
.isEqualTo(range);
assertThat(fetchedRange.statusByReplica())
.hasSize(2)
- .containsEntry("replica1", RestoreRangeStatus.STAGED)
+ .containsEntry("127.0.0.1:12345", RestoreRangeStatus.STAGED)
.containsEntry("replica2", RestoreRangeStatus.CREATED);
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/restore/RestoreJobTestUtils.java b/server/src/test/integration/org/apache/cassandra/sidecar/restore/RestoreJobTestUtils.java
new file mode 100644
index 00000000..cbb35a22
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/restore/RestoreJobTestUtils.java
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.restore;
+
+import java.math.BigInteger;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import com.datastax.driver.core.utils.UUIDs;
+import com.google.inject.AbstractModule;
+import com.google.inject.Module;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.ext.web.client.HttpResponse;
+import io.vertx.ext.web.client.WebClient;
+import org.apache.cassandra.sidecar.cluster.locator.LocalTokenRangesProvider;
+import org.apache.cassandra.sidecar.common.data.ConsistencyLevel;
+import org.apache.cassandra.sidecar.common.request.data.CreateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.config.SidecarConfiguration;
+import org.apache.cassandra.sidecar.db.RestoreRange;
+import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
+import org.apache.cassandra.sidecar.db.schema.SidecarSchema;
+import org.apache.cassandra.sidecar.foundation.RestoreJobSecretsGen;
+import org.apache.cassandra.sidecar.metrics.SidecarMetrics;
+import org.apache.cassandra.sidecar.tasks.ScheduleDecision;
+import org.apache.cassandra.sidecar.utils.SSTableImporter;
+
+import static org.apache.cassandra.testing.utils.AssertionUtils.getBlocking;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test utils for restore job related testing
+ */
+public class RestoreJobTestUtils
+{
+ private RestoreJobTestUtils()
+ {
+ throw new UnsupportedOperationException("Do not instantiate me");
+ }
+
+ public static RestoreJobClient client(WebClient client, String host, int port)
+ {
+ return new RestoreJobClient(client, host, port);
+ }
+
+ public static void assertRestoreRange(RestoreRange range, long startToken, long endToken)
+ {
+ assertRestoreRange(range, startToken, endToken, null);
+ }
+
+ public static void assertRestoreRange(RestoreRange range, long startToken, long endToken, Consumer<RestoreRange> additionalAssertions)
+ {
+ assertThat(range.startToken())
+ .describedAs("startTokens do not match")
+ .isEqualTo(BigInteger.valueOf(startToken));
+ assertThat(range.endToken())
+ .describedAs("endTokens do not match")
+ .isEqualTo(BigInteger.valueOf(endToken));
+ if (additionalAssertions != null)
+ {
+ additionalAssertions.accept(range);
+ }
+ }
+
+ public static UUID createJob(RestoreJobTestUtils.RestoreJobClient testClient, QualifiedTableName tableName)
+ {
+ UUID jobId = UUIDs.timeBased();
+ long expireAt = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(2);
+ CreateRestoreJobRequestPayload payload = CreateRestoreJobRequestPayload
+ .builder(RestoreJobSecretsGen.genRestoreJobSecrets(), expireAt)
+ .jobId(jobId)
+ .consistencyLevel(ConsistencyLevel.LOCAL_QUORUM, "datacenter1").build();
+ testClient.createRestoreJob(tableName, payload);
+ return jobId;
+ }
+
+ public static Module disableRestoreProcessor()
+ {
+ return new AbstractModule()
+ {
+ @Provides
+ @Singleton
+ public RestoreProcessor processor(ExecutorPools executorPools,
+ SidecarConfiguration config,
+ SidecarSchema sidecarSchema,
+ StorageClientPool s3ClientPool,
+ SSTableImporter importer,
+ RestoreRangeDatabaseAccessor rangeDatabaseAccessor,
+ RestoreJobUtil restoreJobUtil,
+ LocalTokenRangesProvider localTokenRangesProvider,
+ SidecarMetrics metrics)
+ {
+ return new RestoreProcessor(executorPools, config, sidecarSchema,
+ s3ClientPool, importer, rangeDatabaseAccessor,
+ restoreJobUtil, localTokenRangesProvider, metrics)
+ {
+ @Override
+ public ScheduleDecision scheduleDecision()
+ {
+ return ScheduleDecision.SKIP;
+ }
+ };
+ }
+ };
+ }
+
+ /**
+ * Wraps the webclient for restore job related API calls
+ */
+ public static class RestoreJobClient
+ {
+ WebClient client;
+ String host;
+ int port;
+
+ private RestoreJobClient(WebClient client, String host, int port)
+ {
+ this.client = client;
+ this.host = host;
+ this.port = port;
+ }
+
+ public void createRestoreJob(QualifiedTableName qtn, CreateRestoreJobRequestPayload payload)
+ {
+ String testRoute = "/api/v1/keyspaces/" + qtn.keyspace() + "/tables/" + qtn.tableName() + "/restore-jobs";
+ HttpResponse<Buffer> response = getBlocking(client.post(port, host, testRoute).sendJson(payload),
+ 10, TimeUnit.SECONDS,
+ "Create RestoreJob");
+ assertThat(response.statusCode()).isEqualTo(HttpResponseStatus.OK.code());
+ }
+
+ public void updateRestoreJob(QualifiedTableName qtn, UUID jobId, UpdateRestoreJobRequestPayload payload)
+ {
+ String testRoute = "/api/v1/keyspaces/" + qtn.keyspace() + "/tables/" + qtn.tableName() + "/restore-jobs/" + jobId;
+ HttpResponse<Buffer> response = getBlocking(client.patch(port, host, testRoute).sendJson(payload),
+ 10, TimeUnit.SECONDS,
+ "Update RestoreJob");
+ assertThat(response.statusCode()).isEqualTo(HttpResponseStatus.OK.code());
+ }
+
+ public void createRestoreSlice(QualifiedTableName qtn, UUID jobId, CreateSliceRequestPayload payload)
+ {
+ String testRoute = "/api/v1/keyspaces/" + qtn.keyspace() + "/tables/" + qtn.tableName() + "/restore-jobs/" + jobId + "/slices";
+ HttpResponse<Buffer> response = getBlocking(client.post(port, host, testRoute).sendJson(payload),
+ 10, TimeUnit.SECONDS,
+ "Create RestoreSlice");
+ assertThat(response.statusCode()).isEqualTo(HttpResponseStatus.CREATED.code());
+ }
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeJoinedIntTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeJoinedIntTest.java
new file mode 100644
index 00000000..23096234
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeJoinedIntTest.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.restore.jobdiscoverer;
+
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.db.RestoreRange;
+import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
+import org.apache.cassandra.sidecar.restore.RestoreJobDiscoverer;
+import org.apache.cassandra.sidecar.restore.RestoreJobTestUtils;
+import org.apache.cassandra.sidecar.restore.RingTopologyRefresher;
+import org.apache.cassandra.sidecar.testing.IntegrationTestBase;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
+import org.apache.cassandra.testing.CassandraIntegrationTest;
+import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
+
+import static org.apache.cassandra.distributed.shared.ClusterUtils.addInstance;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.createJob;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.disableRestoreProcessor;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RestoreJobDiscovererNodeJoinedIntTest extends IntegrationTestBase
+{
+ private static final int NODE = 3;
+ private static final int NODE_JOINED = 4;
+
+ @Override
+ protected void beforeSetup()
+ {
+ installTestSpecificModule(disableRestoreProcessor());
+ }
+
+ @Override
+ protected int[] getInstancesToManage(int clusterSize)
+ {
+ return new int[] { NODE, NODE_JOINED };
+ }
+
+ @CassandraIntegrationTest(nodesPerDc = 3, network = true, buildCluster = false)
+ void testUpdateRestoreRangesWhenNodeMoved(ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ TokenSupplier tokenSupplier = TestTokenSupplier.staticTokens(0, 1000L, 1500L, 2000L);
+ UpgradeableCluster cluster = startCluster(tokenSupplier, cassandraTestContext);
+ RestoreJobTestUtils.RestoreJobClient testClient = RestoreJobTestUtils.client(client, "127.0.0." + NODE, server.actualPort());
+ test(testClient, cluster);
+ }
+
+ private void test(RestoreJobTestUtils.RestoreJobClient testClient, UpgradeableCluster cluster)
+ {
+ // prepare schema
+ waitForSchemaReady(30, TimeUnit.SECONDS);
+ createTestKeyspace(ImmutableMap.of("datacenter1", 2));
+ QualifiedTableName tableName = createTestTable("CREATE TABLE %s (id text PRIMARY KEY, name text);");
+
+ // create job
+ UUID jobId = createJob(testClient, tableName);
+
+ // create slice
+ short bucketId = 0;
+ CreateSliceRequestPayload slicePayload = new CreateSliceRequestPayload("sliceId", bucketId, "bucket", "key",
+ "checksum", BigInteger.valueOf(1L), BigInteger.valueOf(1600L),
+ 100L, 100L);
+ testClient.createRestoreSlice(tableName, jobId, slicePayload);
+
+ // STAGE_READY is required in order to discover slices; update the restore job status
+ testClient.updateRestoreJob(tableName, jobId, UpdateRestoreJobRequestPayload.builder().withStatus(RestoreJobStatus.STAGE_READY).build());
+
+ // first discovery run
+ RestoreJobDiscoverer restoreJobDiscoverer = injector.getInstance(RestoreJobDiscoverer.class);
+ restoreJobDiscoverer.tryExecuteDiscovery();
+
+ RingTopologyRefresher ringTopologyRefresher = injector.getInstance(RingTopologyRefresher.class);
+ Map<Integer, Set<TokenRange>> localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(1)
+ .containsEntry(NODE, ImmutableSet.of(new TokenRange(0, 1000),
+ new TokenRange(1000, 1500)))
+ .doesNotContainKey(NODE_JOINED);
+
+ // assert that no restore ranges are create
+ RestoreRangeDatabaseAccessor rangeDatabaseAccessor = injector.getInstance(RestoreRangeDatabaseAccessor.class);
+ List<RestoreRange> ranges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ Collections.sort(ranges, RestoreRange.TOKEN_BASED_NATURAL_ORDER);
+ assertThat(ranges).hasSize(2);
+ assertThat(ranges.get(0).tokenRange()).isEqualTo(new TokenRange(0, 1000));
+ assertThat(ranges.get(1).tokenRange()).isEqualTo(new TokenRange(1000, 1500));
+
+ // Decommission
+ IUpgradeableInstance seed = cluster.get(1);
+ IUpgradeableInstance joiningNode = addInstance(cluster,
+ seed.config().localDatacenter(),
+ seed.config().localRack(),
+ inst -> {
+ inst.set("auto_bootstrap", true);
+ inst.with(Feature.GOSSIP,
+ Feature.JMX,
+ Feature.NATIVE_PROTOCOL);
+ });
+ joiningNode.startup(cluster);
+ ClusterUtils.awaitRingState(seed, joiningNode, "Normal");
+
+ // Using loopAssert because #onRingTopologyChanged runs in another thread. It takes some time to reflect the RestoreRange update
+ loopAssert(60, 1000, () -> {
+ // Fetch the local token ranges again;
+ // RingTopologyRefresher should detect the topology change and notify RestoreJobDiscover via #onRingTopologyChanged
+ Map<Integer, Set<TokenRange>> currentLocalTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(currentLocalTokenRanges)
+ .hasSize(2)
+ .containsEntry(NODE, ImmutableSet.of(new TokenRange(0, 1000),
+ new TokenRange(1000, 1500)))
+ .containsEntry(NODE_JOINED, ImmutableSet.of(new TokenRange(1000, 1500),
+ new TokenRange(1500, 2000)));
+ List<RestoreRange> restoreRanges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(restoreRanges)
+ .hasSize(3);
+ assertThat(restoreRanges.get(0).tokenRange()).isEqualTo(new TokenRange(0, 1000));
+ assertThat(restoreRanges.get(1).tokenRange()).isEqualTo(new TokenRange(1000, 1500));
+ assertThat(restoreRanges.get(2).tokenRange()).isEqualTo(new TokenRange(1500, 1600));
+ });
+ }
+
+ private static UpgradeableCluster startCluster(TokenSupplier tokenSupplier, ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ return cassandraTestContext.configureAndStartCluster(builder -> {
+ builder.withTokenSupplier(tokenSupplier);
+ });
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeJoiningIntTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeJoiningIntTest.java
new file mode 100644
index 00000000..9b969de1
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeJoiningIntTest.java
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.restore.jobdiscoverer;
+
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.db.RestoreRange;
+import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
+import org.apache.cassandra.sidecar.restore.RestoreJobDiscoverer;
+import org.apache.cassandra.sidecar.restore.RestoreJobTestUtils;
+import org.apache.cassandra.sidecar.restore.RingTopologyRefresher;
+import org.apache.cassandra.sidecar.testing.IntegrationTestBase;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
+import org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperJoiningNode;
+import org.apache.cassandra.testing.CassandraIntegrationTest;
+import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
+
+import static org.apache.cassandra.distributed.shared.ClusterUtils.addInstance;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.assertRestoreRange;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.createJob;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.disableRestoreProcessor;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RestoreJobDiscovererNodeJoiningIntTest extends IntegrationTestBase
+{
+ private static final int MANAGED_CASSANDRA_NODE_NUM = 2;
+ private static final String MANAGED_CASSANDRA_NODE_IP = "127.0.0." + MANAGED_CASSANDRA_NODE_NUM;
+
+ @Override
+ protected void beforeSetup()
+ {
+ installTestSpecificModule(disableRestoreProcessor());
+ }
+
+ @CassandraIntegrationTest(nodesPerDc = 3, newNodesPerDc = 1, network = true, buildCluster = false)
+ void testUpdateRestoreRangesWhenNewNodeJoining(ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ TokenSupplier tokenSupplier = TestTokenSupplier.staticTokens(0, 1000L, 1500L, 2000L);
+ UpgradeableCluster cluster = startCluster(tokenSupplier, cassandraTestContext, 4);
+ try
+ {
+ test(BBHelperJoiningNode.transientStateStart, cluster);
+ }
+ finally
+ {
+ BBHelperJoiningNode.transientStateEnd.countDown();
+ }
+ }
+
+ @Override
+ protected int[] getInstancesToManage(int clusterSize)
+ {
+ return new int[] { MANAGED_CASSANDRA_NODE_NUM };
+ }
+
+ private void test(CountDownLatch transientStateStart, UpgradeableCluster cluster)
+ {
+ // prepare schema
+ waitForSchemaReady(30, TimeUnit.SECONDS);
+ createTestKeyspace(ImmutableMap.of("datacenter1", 2));
+ QualifiedTableName tableName = createTestTable("CREATE TABLE %s (id text PRIMARY KEY, name text);");
+
+ // create job
+ RestoreJobTestUtils.RestoreJobClient testClient = RestoreJobTestUtils.client(client, MANAGED_CASSANDRA_NODE_IP, server.actualPort());
+ UUID jobId = createJob(testClient, tableName);
+
+ // create slice
+ short bucketId = 0;
+ CreateSliceRequestPayload slicePayload = new CreateSliceRequestPayload("sliceId", bucketId, "bucket", "key",
+ "checksum", BigInteger.valueOf(1001L), BigInteger.valueOf(1600L),
+ 100L, 100L);
+ testClient.createRestoreSlice(tableName, jobId, slicePayload);
+
+ // STAGE_READY is required in order to discover slices; update the restore job status
+ testClient.updateRestoreJob(tableName, jobId, UpdateRestoreJobRequestPayload.builder().withStatus(RestoreJobStatus.STAGE_READY).build());
+
+ // first discovery run
+ RestoreJobDiscoverer restoreJobDiscoverer = injector.getInstance(RestoreJobDiscoverer.class);
+ restoreJobDiscoverer.tryExecuteDiscovery();
+
+ RingTopologyRefresher ringTopologyRefresher = injector.getInstance(RingTopologyRefresher.class);
+ Map<Integer, Set<TokenRange>> localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(1)
+ .containsEntry(MANAGED_CASSANDRA_NODE_NUM,
+ ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1000),
+ new TokenRange(1500, Long.MAX_VALUE)));
+
+ // assert that no restore ranges are create
+ RestoreRangeDatabaseAccessor rangeDatabaseAccessor = injector.getInstance(RestoreRangeDatabaseAccessor.class);
+ List<RestoreRange> ranges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(ranges).hasSize(1);
+ assertRestoreRange(ranges.get(0), 1500L, 1600L);
+
+ // start move in the background
+ IUpgradeableInstance seed = cluster.get(1);
+ IUpgradeableInstance joiningNode = addInstance(cluster,
+ seed.config().localDatacenter(),
+ seed.config().localRack(),
+ inst -> {
+ inst.set("auto_bootstrap", true);
+ inst.with(Feature.GOSSIP,
+ Feature.JMX,
+ Feature.NATIVE_PROTOCOL);
+ });
+ startAsync("Start new node node" + joiningNode.config().num(),
+ () -> joiningNode.startup(cluster));
+
+ // Wait until nodes have reached expected state
+ awaitLatchOrThrow(transientStateStart, 2, TimeUnit.MINUTES, "transientStateStart");
+ ClusterUtils.awaitRingState(seed, joiningNode, "Joining");
+
+ // Fetch the local token ranges again;
+ // RingTopologyRefresher should detect the topology change and notify RestoreJobDiscover via #onRingTopologyChanged
+ localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .describedAs("There is a new node of 2000 joining. " +
+ "Node 2 now has range (Long.MIN, 0], (0, 1000], (1500, 2000] and (2000, Long.MAX].")
+ .hasSize(1)
+ .containsEntry(MANAGED_CASSANDRA_NODE_NUM,
+ ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1000),
+ // (1500, 2000] remains temporarily while new node is joining; once node joined, the range is no longer owned
+ new TokenRange(1500, 2000),
+ new TokenRange(2000, Long.MAX_VALUE)));
+
+ // Using loopAssert because #onRingTopologyChanged runs in another thread. It takes some time to reflect the RestoreRange update
+ loopAssert(10, 500, () -> {
+ List<RestoreRange> restoreRanges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(restoreRanges)
+ .describedAs("Local token ranges are effectively the same. Therefore restore ranges do not change")
+ .hasSize(1);
+ assertRestoreRange(ranges.get(0), 1500L, 1600L);
+ });
+ }
+
+ @Override
+ protected String subjectAlternativeNameIpAddress()
+ {
+ return MANAGED_CASSANDRA_NODE_IP;
+ }
+
+ private static UpgradeableCluster startCluster(TokenSupplier tokenSupplier,
+ ConfigurableCassandraTestContext cassandraTestContext,
+ int joiningNodeNum)
+ {
+ BBHelperJoiningNode.reset();
+ return cassandraTestContext.configureAndStartCluster(builder -> {
+ builder.withInstanceInitializer((cl, num) -> BBHelperJoiningNode.install(cl, num, joiningNodeNum));
+ builder.withTokenSupplier(tokenSupplier);
+ });
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeLeavingIntTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeLeavingIntTest.java
new file mode 100644
index 00000000..b350f3ea
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeLeavingIntTest.java
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.restore.jobdiscoverer;
+
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.db.RestoreRange;
+import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
+import org.apache.cassandra.sidecar.restore.RestoreJobDiscoverer;
+import org.apache.cassandra.sidecar.restore.RestoreJobTestUtils;
+import org.apache.cassandra.sidecar.restore.RingTopologyRefresher;
+import org.apache.cassandra.sidecar.testing.IntegrationTestBase;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
+import org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperLeavingNode;
+import org.apache.cassandra.testing.CassandraIntegrationTest;
+import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
+
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.assertRestoreRange;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.createJob;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.disableRestoreProcessor;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RestoreJobDiscovererNodeLeavingIntTest extends IntegrationTestBase
+{
+ private static final int MANAGED_CASSANDRA_NODE_NUM = 2;
+ private static final String MANAGED_CASSANDRA_NODE_IP = "127.0.0." + MANAGED_CASSANDRA_NODE_NUM;
+
+ @Override
+ protected void beforeSetup()
+ {
+ installTestSpecificModule(disableRestoreProcessor());
+ }
+
+ @CassandraIntegrationTest(nodesPerDc = 3, network = true, buildCluster = false)
+ void testUpdateRestoreRangesWhenNodeLeaving(ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ TokenSupplier tokenSupplier = TestTokenSupplier.staticTokens(0, 1000L, 2000L);
+ int leavingNodeNum = 3;
+ UpgradeableCluster cluster = startCluster(tokenSupplier, cassandraTestContext, leavingNodeNum);
+ try
+ {
+ test(BBHelperLeavingNode.transientStateStart, cluster, leavingNodeNum);
+ }
+ finally
+ {
+ BBHelperLeavingNode.transientStateEnd.countDown();
+ }
+ }
+
+ @Override
+ protected int[] getInstancesToManage(int clusterSize)
+ {
+ return new int[] { MANAGED_CASSANDRA_NODE_NUM };
+ }
+
+ private void test(CountDownLatch transientStateStart, UpgradeableCluster cluster, int leavingNodeNum)
+ {
+ // prepare schema
+ waitForSchemaReady(30, TimeUnit.SECONDS);
+ createTestKeyspace(ImmutableMap.of("datacenter1", 2));
+ QualifiedTableName tableName = createTestTable("CREATE TABLE %s (id text PRIMARY KEY, name text);");
+
+ // create job
+ RestoreJobTestUtils.RestoreJobClient testClient = RestoreJobTestUtils.client(client, MANAGED_CASSANDRA_NODE_IP, server.actualPort());
+ UUID jobId = createJob(testClient, tableName);
+
+ // create slice
+ short bucketId = 0;
+ CreateSliceRequestPayload slicePayload = new CreateSliceRequestPayload("sliceId", bucketId, "bucket", "key",
+ "checksum", BigInteger.valueOf(1001L), BigInteger.valueOf(1600L),
+ 100L, 100L);
+ testClient.createRestoreSlice(tableName, jobId, slicePayload);
+
+ // STAGE_READY is required in order to discover slices; update the restore job status
+ testClient.updateRestoreJob(tableName, jobId, UpdateRestoreJobRequestPayload.builder().withStatus(RestoreJobStatus.STAGE_READY).build());
+
+ // first discovery run
+ RestoreJobDiscoverer restoreJobDiscoverer = injector.getInstance(RestoreJobDiscoverer.class);
+ restoreJobDiscoverer.tryExecuteDiscovery();
+
+ RingTopologyRefresher ringTopologyRefresher = injector.getInstance(RingTopologyRefresher.class);
+ Map<Integer, Set<TokenRange>> localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(1)
+ .containsEntry(MANAGED_CASSANDRA_NODE_NUM,
+ ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1000),
+ new TokenRange(2000, Long.MAX_VALUE)));
+
+ // assert that no restore ranges are create
+ RestoreRangeDatabaseAccessor rangeDatabaseAccessor = injector.getInstance(RestoreRangeDatabaseAccessor.class);
+ List<RestoreRange> ranges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(ranges)
+ .describedAs("No RestoreRange is created because no slice overlaps with local token ranges")
+ .isEmpty();
+
+ // start move in the background
+ IUpgradeableInstance seed = cluster.get(1);
+ IUpgradeableInstance node = cluster.get(leavingNodeNum);
+ startAsync("Decommission node" + node.config().num(),
+ // testing keyspace has RF == 2. Using --force does not hurt fault tolerance.
+ () -> node.nodetoolResult("decommission", "--force").asserts().success());
+
+ // Wait until nodes have reached expected state
+ awaitLatchOrThrow(transientStateStart, 2, TimeUnit.MINUTES, "transientStateStart");
+ ClusterUtils.awaitRingState(seed, node, "Leaving");
+
+ // Fetch the local token ranges again;
+ // RingTopologyRefresher should detect the topology change and notify RestoreJobDiscover via #onRingTopologyChanged
+ localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(1)
+ .containsEntry(MANAGED_CASSANDRA_NODE_NUM,
+ ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1000),
+ new TokenRange(1000, 2000), // gained due to node leaving; now it own the entire ring effectively.
+ new TokenRange(2000, Long.MAX_VALUE)));
+
+ // Using loopAssert because #onRingTopologyChanged runs in another thread. It takes some time to reflect the RestoreRange update
+ loopAssert(10, 500, () -> {
+ List<RestoreRange> restoreRanges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(restoreRanges)
+ .describedAs("A restore range should be created. After the topology change, now the slice is partially owned by the local node")
+ .hasSize(1);
+ assertRestoreRange(restoreRanges.get(0), 1000L, 1600L);
+ });
+ }
+
+ @Override
+ protected String subjectAlternativeNameIpAddress()
+ {
+ return MANAGED_CASSANDRA_NODE_IP;
+ }
+
+ private static UpgradeableCluster startCluster(TokenSupplier tokenSupplier, ConfigurableCassandraTestContext cassandraTestContext, int leavingNodeNum)
+ {
+ BBHelperLeavingNode.reset();
+ return cassandraTestContext.configureAndStartCluster(builder -> {
+ builder.withInstanceInitializer((cl, num) -> BBHelperLeavingNode.install(cl, num, leavingNodeNum));
+ builder.withTokenSupplier(tokenSupplier);
+ });
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeLeftIntTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeLeftIntTest.java
new file mode 100644
index 00000000..0e9fb474
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeLeftIntTest.java
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.restore.jobdiscoverer;
+
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.db.RestoreRange;
+import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
+import org.apache.cassandra.sidecar.restore.RestoreJobDiscoverer;
+import org.apache.cassandra.sidecar.restore.RestoreJobTestUtils;
+import org.apache.cassandra.sidecar.restore.RingTopologyRefresher;
+import org.apache.cassandra.sidecar.testing.IntegrationTestBase;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
+import org.apache.cassandra.testing.CassandraIntegrationTest;
+import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
+
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.createJob;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.disableRestoreProcessor;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RestoreJobDiscovererNodeLeftIntTest extends IntegrationTestBase
+{
+ private static final int NODE_1 = 1;
+ private static final int NODE_LEFT = 2;
+
+ @Override
+ protected void beforeSetup()
+ {
+ installTestSpecificModule(disableRestoreProcessor());
+ }
+
+ @Override
+ protected int[] getInstancesToManage(int clusterSize)
+ {
+ return new int[] { NODE_1, NODE_LEFT };
+ }
+
+ @CassandraIntegrationTest(nodesPerDc = 3, network = true, buildCluster = false)
+ void testUpdateRestoreRangesWhenNodeMoved(ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ TokenSupplier tokenSupplier = TestTokenSupplier.staticTokens(0, 1000L, 2000L);
+ UpgradeableCluster cluster = startCluster(tokenSupplier, cassandraTestContext);
+ RestoreJobTestUtils.RestoreJobClient testClient = RestoreJobTestUtils.client(client, "127.0.0.1", server.actualPort());
+ test(testClient, cluster);
+ }
+
+ private void test(RestoreJobTestUtils.RestoreJobClient testClient, UpgradeableCluster cluster)
+ {
+ // prepare schema
+ waitForSchemaReady(30, TimeUnit.SECONDS);
+ createTestKeyspace(ImmutableMap.of("datacenter1", 2));
+ QualifiedTableName tableName = createTestTable("CREATE TABLE %s (id text PRIMARY KEY, name text);");
+
+ // create job
+ UUID jobId = createJob(testClient, tableName);
+
+ // create slice
+ short bucketId = 0;
+ CreateSliceRequestPayload slicePayload = new CreateSliceRequestPayload("sliceId", bucketId, "bucket", "key",
+ "checksum", BigInteger.valueOf(1L), BigInteger.valueOf(1600L),
+ 100L, 100L);
+ testClient.createRestoreSlice(tableName, jobId, slicePayload);
+
+ // STAGE_READY is required in order to discover slices; update the restore job status
+ testClient.updateRestoreJob(tableName, jobId, UpdateRestoreJobRequestPayload.builder().withStatus(RestoreJobStatus.STAGE_READY).build());
+
+ // first discovery run
+ RestoreJobDiscoverer restoreJobDiscoverer = injector.getInstance(RestoreJobDiscoverer.class);
+ restoreJobDiscoverer.tryExecuteDiscovery();
+
+ RingTopologyRefresher ringTopologyRefresher = injector.getInstance(RingTopologyRefresher.class);
+ Map<Integer, Set<TokenRange>> localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(2)
+ .containsEntry(NODE_1, ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(1000, 2000),
+ new TokenRange(2000, Long.MAX_VALUE)))
+ .containsEntry(NODE_LEFT, ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1000),
+ new TokenRange(2000, Long.MAX_VALUE)));
+
+ // assert that no restore ranges are create
+ RestoreRangeDatabaseAccessor rangeDatabaseAccessor = injector.getInstance(RestoreRangeDatabaseAccessor.class);
+ List<RestoreRange> ranges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ Collections.sort(ranges, RestoreRange.TOKEN_BASED_NATURAL_ORDER);
+ assertThat(ranges).hasSize(2);
+ assertThat(ranges.get(0).tokenRange()).isEqualTo(new TokenRange(0, 1000)); // node 2
+ assertThat(ranges.get(1).tokenRange()).isEqualTo(new TokenRange(1000, 1600)); // node 1
+
+ // Decommission
+ IUpgradeableInstance node = cluster.get(NODE_LEFT);
+ // testing keyspace has RF == 2. Using --force does not hurt fault tolerance.
+ node.nodetoolResult("decommission", "--force").asserts().success();
+
+ // Fetch the local token ranges again;
+ // RingTopologyRefresher should detect the topology change and notify RestoreJobDiscover via #onRingTopologyChanged
+ localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(1)
+ .containsEntry(NODE_1, ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 2000), // expanded from (1000, 2000] to (0, 2000] due to left node
+ new TokenRange(2000, Long.MAX_VALUE)))
+ .doesNotContainKey(NODE_LEFT);
+
+ // Using loopAssert because #onRingTopologyChanged runs in another thread. It takes some time to reflect the RestoreRange update
+ loopAssert(30, 1000, () -> {
+ List<RestoreRange> restoreRanges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(restoreRanges)
+ .hasSize(2);
+ assertThat(restoreRanges.get(0).tokenRange()).isEqualTo(new TokenRange(0, 1000)); // node 1 now gains the range and create the restoreRange
+ assertThat(restoreRanges.get(1).tokenRange()).isEqualTo(new TokenRange(1000, 1600));
+ });
+ }
+
+ private static UpgradeableCluster startCluster(TokenSupplier tokenSupplier, ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ return cassandraTestContext.configureAndStartCluster(builder -> {
+ builder.withTokenSupplier(tokenSupplier);
+ });
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeMovedIntTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeMovedIntTest.java
new file mode 100644
index 00000000..8f19acc9
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeMovedIntTest.java
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.restore.jobdiscoverer;
+
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus;
+import org.apache.cassandra.sidecar.common.server.utils.StringUtils;
+import org.apache.cassandra.sidecar.db.RestoreRange;
+import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
+import org.apache.cassandra.sidecar.restore.RestoreJobDiscoverer;
+import org.apache.cassandra.sidecar.restore.RestoreJobTestUtils;
+import org.apache.cassandra.sidecar.restore.RingTopologyRefresher;
+import org.apache.cassandra.sidecar.testing.IntegrationTestBase;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
+import org.apache.cassandra.testing.CassandraIntegrationTest;
+import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
+
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.assertRestoreRange;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.createJob;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.disableRestoreProcessor;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test the scenario that node has moved token while restore job is in progress
+ */
+class RestoreJobDiscovererNodeMovedIntTest extends IntegrationTestBase
+{
+ private static final int NODE_1 = 1;
+ private static final int NODE_2 = 2;
+
+ @Override
+ protected void beforeSetup()
+ {
+ installTestSpecificModule(disableRestoreProcessor());
+ }
+
+ @Override
+ protected int[] getInstancesToManage(int clusterSize)
+ {
+ return new int[] { 1, 2 };
+ }
+
+ @CassandraIntegrationTest(nodesPerDc = 3, network = true, buildCluster = false)
+ void testUpdateRestoreRangesWhenNodeMoved(ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ TokenSupplier tokenSupplier = TestTokenSupplier.staticTokens(0, 1000L, 2000L);
+ UpgradeableCluster cluster = startCluster(tokenSupplier, cassandraTestContext);
+ long newToken = 1500L;
+ RestoreJobTestUtils.RestoreJobClient testClient = RestoreJobTestUtils.client(client, "127.0.0.1", server.actualPort());
+ test(testClient, cluster, newToken);
+ }
+
+ private void test(RestoreJobTestUtils.RestoreJobClient testClient, UpgradeableCluster cluster, long moveTargetToken)
+ {
+ // prepare schema
+ waitForSchemaReady(30, TimeUnit.SECONDS);
+ createTestKeyspace(ImmutableMap.of("datacenter1", 2));
+ QualifiedTableName tableName = createTestTable("CREATE TABLE %s (id text PRIMARY KEY, name text);");
+
+ // create job
+ UUID jobId = createJob(testClient, tableName);
+
+ // create slice
+ short bucketId = 0;
+ CreateSliceRequestPayload slicePayload = new CreateSliceRequestPayload("sliceId", bucketId, "bucket", "key",
+ "checksum", BigInteger.valueOf(1001L), BigInteger.valueOf(1600L),
+ 100L, 100L);
+ testClient.createRestoreSlice(tableName, jobId, slicePayload);
+
+ // STAGE_READY is required in order to discover slices; update the restore job status
+ testClient.updateRestoreJob(tableName, jobId, UpdateRestoreJobRequestPayload.builder().withStatus(RestoreJobStatus.STAGE_READY).build());
+
+ // first discovery run
+ RestoreJobDiscoverer restoreJobDiscoverer = injector.getInstance(RestoreJobDiscoverer.class);
+ restoreJobDiscoverer.tryExecuteDiscovery();
+
+ RingTopologyRefresher ringTopologyRefresher = injector.getInstance(RingTopologyRefresher.class);
+ Map<Integer, Set<TokenRange>> localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(2)
+ .containsEntry(NODE_1, ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(1000, 2000),
+ new TokenRange(2000, Long.MAX_VALUE)))
+ .containsEntry(NODE_2, ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1000),
+ new TokenRange(2000, Long.MAX_VALUE)));
+
+ // assert that no restore ranges are create
+ RestoreRangeDatabaseAccessor rangeDatabaseAccessor = injector.getInstance(RestoreRangeDatabaseAccessor.class);
+ List<RestoreRange> ranges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(ranges)
+ .describedAs("node 1 and only node 1 should create the restore range")
+ .hasSize(1);
+ assertThat(ranges.get(0).tokenRange()).isEqualTo(new TokenRange(1000, 1600));
+
+ // Move token
+ IUpgradeableInstance movingNode = cluster.get(2);
+ movingNode.nodetoolResult("move", "--", Long.toString(moveTargetToken))
+ .asserts()
+ .success();
+ // Wait until nodes have reached expected state
+ ClusterUtils.awaitRingState(cluster.get(1), movingNode, "Normal");
+ ClusterUtils.awaitRingState(cluster.get(3), movingNode, "Normal");
+
+ // Fetch the local token ranges again;
+ // RingTopologyRefresher should detect the topology change and notify RestoreJobDiscover via #onRingTopologyChanged
+ localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(2)
+ .containsEntry(NODE_1, ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(1500, 2000),
+ new TokenRange(2000, Long.MAX_VALUE)))
+ .containsEntry(NODE_2, ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1500),
+ new TokenRange(2000, Long.MAX_VALUE)));
+
+ // Using loopAssert because #onRingTopologyChanged runs in another thread. It takes some time to reflect the RestoreRange update
+ loopAssert(30, 1000, () -> {
+ List<RestoreRange> restoreRanges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(restoreRanges)
+ .describedAs("A restore range should be created. After the topology change, now the slice is partially owned by the local node")
+ .hasSize(3);
+ Collections.sort(restoreRanges, RestoreRange.TOKEN_BASED_NATURAL_ORDER);
+ assertRestoreRange(restoreRanges.get(0), 1000L, 1500L); // in node 2
+ assertRestoreRange(restoreRanges.get(1), 1000L, 1600L, range -> {
+ assertThat(range.statusByReplica())
+ .describedAs("The original restore range from node 1 should be discarded, due to losing the range ownership")
+ .hasSize(1)
+ .containsEntry(StringUtils.cassandraFormattedHostAndPort(cluster.get(1).broadcastAddress()), RestoreRangeStatus.DISCARDED);
+ });
+ assertRestoreRange(restoreRanges.get(2), 1500L, 1600L); // in node 1
+ });
+ }
+
+ private static UpgradeableCluster startCluster(TokenSupplier tokenSupplier, ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ return cassandraTestContext.configureAndStartCluster(builder -> {
+ builder.withTokenSupplier(tokenSupplier);
+ });
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeMovingIntTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeMovingIntTest.java
new file mode 100644
index 00000000..d8742002
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/restore/jobdiscoverer/RestoreJobDiscovererNodeMovingIntTest.java
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.restore.jobdiscoverer;
+
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.db.RestoreRange;
+import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
+import org.apache.cassandra.sidecar.restore.RestoreJobDiscoverer;
+import org.apache.cassandra.sidecar.restore.RestoreJobTestUtils;
+import org.apache.cassandra.sidecar.restore.RingTopologyRefresher;
+import org.apache.cassandra.sidecar.testing.IntegrationTestBase;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
+import org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperMovingNode;
+import org.apache.cassandra.testing.CassandraIntegrationTest;
+import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
+
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.assertRestoreRange;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.createJob;
+import static org.apache.cassandra.sidecar.restore.RestoreJobTestUtils.disableRestoreProcessor;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test the scenario that token move is triggered while restore job is in progress
+ */
+class RestoreJobDiscovererNodeMovingIntTest extends IntegrationTestBase
+{
+ private static final int MANAGED_CASSANDRA_NODE_NUM = 2;
+ private static final String MANAGED_CASSANDRA_NODE_IP = "127.0.0." + MANAGED_CASSANDRA_NODE_NUM;
+
+ @Override
+ protected void beforeSetup()
+ {
+ installTestSpecificModule(disableRestoreProcessor());
+ }
+
+ @CassandraIntegrationTest(nodesPerDc = 3, network = true, buildCluster = false)
+ void testUpdateRestoreRangesWhenNodeMoving(ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ TokenSupplier tokenSupplier = TestTokenSupplier.staticTokens(0, 1000L, 2000L);
+ UpgradeableCluster cluster = startCluster(tokenSupplier, cassandraTestContext);
+ long newToken = 1500L;
+ try
+ {
+ test(BBHelperMovingNode.transientStateStart, cluster, newToken);
+ }
+ finally
+ {
+ BBHelperMovingNode.transientStateEnd.countDown();
+ }
+ }
+
+ @Override
+ protected int[] getInstancesToManage(int clusterSize)
+ {
+ return new int[] { MANAGED_CASSANDRA_NODE_NUM };
+ }
+
+ private void test(CountDownLatch transientStateStart, UpgradeableCluster cluster, long moveTargetToken)
+ {
+ // prepare schema
+ waitForSchemaReady(30, TimeUnit.SECONDS);
+ createTestKeyspace(ImmutableMap.of("datacenter1", 2));
+ QualifiedTableName tableName = createTestTable("CREATE TABLE %s (id text PRIMARY KEY, name text);");
+
+ // create job
+ RestoreJobTestUtils.RestoreJobClient testClient = RestoreJobTestUtils.client(client, MANAGED_CASSANDRA_NODE_IP, server.actualPort());
+ UUID jobId = createJob(testClient, tableName);
+
+ // create slice
+ short bucketId = 0;
+ CreateSliceRequestPayload slicePayload = new CreateSliceRequestPayload("sliceId", bucketId, "bucket", "key",
+ "checksum", BigInteger.valueOf(1001L), BigInteger.valueOf(1600L),
+ 100L, 100L);
+ testClient.createRestoreSlice(tableName, jobId, slicePayload);
+
+ // STAGE_READY is required in order to discover slices; update the restore job status
+ testClient.updateRestoreJob(tableName, jobId, UpdateRestoreJobRequestPayload.builder().withStatus(RestoreJobStatus.STAGE_READY).build());
+
+ // first discovery run
+ RestoreJobDiscoverer restoreJobDiscoverer = injector.getInstance(RestoreJobDiscoverer.class);
+ restoreJobDiscoverer.tryExecuteDiscovery();
+
+ RingTopologyRefresher ringTopologyRefresher = injector.getInstance(RingTopologyRefresher.class);
+ Map<Integer, Set<TokenRange>> localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(1)
+ .containsEntry(MANAGED_CASSANDRA_NODE_NUM,
+ ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1000),
+ new TokenRange(2000, Long.MAX_VALUE)));
+
+ // assert that no restore ranges are create
+ RestoreRangeDatabaseAccessor rangeDatabaseAccessor = injector.getInstance(RestoreRangeDatabaseAccessor.class);
+ List<RestoreRange> ranges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(ranges)
+ .describedAs("No RestoreRange is created because no slice overlaps with local token ranges")
+ .isEmpty();
+
+ // start move in the background
+ IUpgradeableInstance seed = cluster.get(1);
+ IUpgradeableInstance movingNode = cluster.get(MANAGED_CASSANDRA_NODE_NUM);
+ startAsync("move token of node" + movingNode.config().num() + " to " + moveTargetToken,
+ () -> movingNode.nodetoolResult("move", "--", Long.toString(moveTargetToken))
+ .asserts()
+ .success());
+
+ // Wait until nodes have reached expected state
+ awaitLatchOrThrow(transientStateStart, 2, TimeUnit.MINUTES, "transientStateStart");
+ ClusterUtils.awaitRingState(seed, movingNode, "Moving");
+
+ // Fetch the local token ranges again;
+ // RingTopologyRefresher should detect the topology change and notify RestoreJobDiscover via #onRingTopologyChanged
+ localTokenRanges = ringTopologyRefresher.localTokenRanges(tableName.keyspace(), true);
+ assertThat(localTokenRanges)
+ .hasSize(1)
+ .containsEntry(MANAGED_CASSANDRA_NODE_NUM,
+ ImmutableSet.of(new TokenRange(Long.MIN_VALUE, 0),
+ new TokenRange(0, 1000),
+ new TokenRange(1000, 1500), // range (1000, 1500] is gained due to move
+ new TokenRange(2000, Long.MAX_VALUE)));
+
+ // Using loopAssert because #onRingTopologyChanged runs in another thread. It takes some time to reflect the RestoreRange update
+ loopAssert(10, 500, () -> {
+ List<RestoreRange> restoreRanges = rangeDatabaseAccessor.findAll(jobId, bucketId);
+ assertThat(restoreRanges)
+ .describedAs("A restore range should be created. After the topology change, now the slice is partially owned by the local node")
+ .hasSize(1);
+ assertRestoreRange(restoreRanges.get(0), 1000L, 1500L);
+ });
+ }
+
+ @Override
+ protected String subjectAlternativeNameIpAddress()
+ {
+ return MANAGED_CASSANDRA_NODE_IP;
+ }
+
+ private static UpgradeableCluster startCluster(TokenSupplier tokenSupplier, ConfigurableCassandraTestContext cassandraTestContext)
+ {
+ BBHelperMovingNode.reset();
+ return cassandraTestContext.configureAndStartCluster(builder -> {
+ builder.withInstanceInitializer((cl, num) -> BBHelperMovingNode.install(cl, num, 2));
+ builder.withTokenSupplier(tokenSupplier);
+ });
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/BaseTokenRangeIntegrationTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/BaseTokenRangeIntegrationTest.java
index a05e4d21..bbca5155 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/BaseTokenRangeIntegrationTest.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/BaseTokenRangeIntegrationTest.java
@@ -43,6 +43,7 @@ import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
import org.apache.cassandra.sidecar.common.server.cluster.locator.Partitioners;
import org.apache.cassandra.sidecar.testing.IntegrationTestBase;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
import org.apache.cassandra.testing.AbstractCassandraTestContext;
import org.apache.cassandra.testing.CassandraIntegrationTest;
import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningBaseTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningBaseTest.java
index 5dad7933..c6ad977b 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningBaseTest.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningBaseTest.java
@@ -44,6 +44,7 @@ import org.apache.cassandra.distributed.api.IUpgradeableInstance;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
import org.apache.cassandra.testing.CassandraIntegrationTest;
import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestDoubleCluster.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestDoubleCluster.java
index b550a9c0..5e20af72 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestDoubleCluster.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestDoubleCluster.java
@@ -176,7 +176,7 @@ public class JoiningTestDoubleCluster extends JoiningBaseTest
// We intercept the bootstrap of the new nodes (6-10) to validate token ranges
if (nodeNumber > 5)
{
- BootstrapBBUtils.installSetBoostrapStateIntercepter(cl, BBHelperDoubleClusterSize.class);
+ BootstrapBBUtils.installSetBoostrapStateInterceptor(cl, BBHelperDoubleClusterSize.class);
}
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultiDC.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultiDC.java
index bff3ca50..34dfb981 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultiDC.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultiDC.java
@@ -232,7 +232,7 @@ public class JoiningTestMultiDC extends JoiningBaseTest
// We intercept the bootstrap of nodes (7-12) to validate token ranges
if (nodeNumber > 6)
{
- BootstrapBBUtils.installSetBoostrapStateIntercepter(cl, BBHelperDoubleClusterMultiDC.class);
+ BootstrapBBUtils.installSetBoostrapStateInterceptor(cl, BBHelperDoubleClusterMultiDC.class);
}
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultiDCSingleReplicated.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultiDCSingleReplicated.java
index 33ecf0f3..5610986f 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultiDCSingleReplicated.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultiDCSingleReplicated.java
@@ -161,7 +161,7 @@ public class JoiningTestMultiDCSingleReplicated extends JoiningBaseTest
// We intercept the bootstrap of nodes (11,12) to validate token ranges
if (nodeNumber > 10)
{
- BootstrapBBUtils.installSetBoostrapStateIntercepter(cl, BBHelperMultiDC.class);
+ BootstrapBBUtils.installSetBoostrapStateInterceptor(cl, BBHelperMultiDC.class);
}
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultipleNodes.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultipleNodes.java
index 31da33a7..9a027cd6 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultipleNodes.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestMultipleNodes.java
@@ -148,7 +148,7 @@ public class JoiningTestMultipleNodes extends JoiningBaseTest
// We intercept the joining of nodes (4, 5) to validate token ranges
if (nodeNumber > 3)
{
- BootstrapBBUtils.installSetBoostrapStateIntercepter(cl, BBHelperMultipleJoiningNodes.class);
+ BootstrapBBUtils.installSetBoostrapStateInterceptor(cl, BBHelperMultipleJoiningNodes.class);
}
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestSingleNode.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestSingleNode.java
index 03bed029..1e574287 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestSingleNode.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/JoiningTestSingleNode.java
@@ -23,18 +23,13 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
import com.google.common.collect.Range;
import org.junit.jupiter.api.extension.ExtendWith;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
-import net.bytebuddy.implementation.bind.annotation.SuperCall;
-import org.apache.cassandra.db.SystemKeyspace;
-import org.apache.cassandra.sidecar.testing.BootstrapBBUtils;
+import org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperJoiningNode;
import org.apache.cassandra.testing.CassandraIntegrationTest;
import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
@@ -52,12 +47,14 @@ public class JoiningTestSingleNode extends JoiningBaseTest
void retrieveMappingWithJoiningNode(VertxTestContext context,
ConfigurableCassandraTestContext cassandraTestContext) throws Exception
{
- BBHelperSingleJoiningNode.reset();
+ BBHelperJoiningNode.reset();
runJoiningTestScenario(context,
cassandraTestContext,
- BBHelperSingleJoiningNode::install,
- BBHelperSingleJoiningNode.transientStateStart,
- BBHelperSingleJoiningNode.transientStateEnd,
+ // Test case involves 5 node cluster with 1 joining node
+ // We intercept the bootstrap of the joining node (6) to validate token ranges
+ (cl, num) -> BBHelperJoiningNode.install(cl, num, 6),
+ BBHelperJoiningNode.transientStateStart,
+ BBHelperJoiningNode.transientStateEnd,
generateExpectedRangeMappingSingleJoiningNode());
}
@@ -98,40 +95,4 @@ public class JoiningTestSingleNode extends JoiningBaseTest
}
};
}
-
- /**
- * ByteBuddy helper for a single joining node
- */
- public static class BBHelperSingleJoiningNode
- {
- static CountDownLatch transientStateStart = new CountDownLatch(1);
- static CountDownLatch transientStateEnd = new CountDownLatch(1);
-
- public static void install(ClassLoader cl, Integer nodeNumber)
- {
- // Test case involves 3 node cluster with 1 joining node
- // We intercept the bootstrap of the leaving node (4) to validate token ranges
- if (nodeNumber == 6)
- {
- BootstrapBBUtils.installSetBoostrapStateIntercepter(cl, BBHelperSingleJoiningNode.class);
- }
- }
-
- public static void setBootstrapState(SystemKeyspace.BootstrapState state, @SuperCall Callable<Void> orig) throws Exception
- {
- if (state == SystemKeyspace.BootstrapState.COMPLETED)
- {
- // trigger bootstrap start and wait until bootstrap is ready from test
- transientStateStart.countDown();
- awaitLatchOrTimeout(transientStateEnd, 2, TimeUnit.MINUTES, "transientStateEnd");
- }
- orig.call();
- }
-
- public static void reset()
- {
- transientStateStart = new CountDownLatch(1);
- transientStateEnd = new CountDownLatch(1);
- }
- }
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/LeavingTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/LeavingTest.java
index 0743aa4f..63f32cb8 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/LeavingTest.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/LeavingTest.java
@@ -44,6 +44,8 @@ import net.bytebuddy.implementation.bind.annotation.SuperCall;
import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
+import org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperLeavingNode;
import org.apache.cassandra.testing.CassandraIntegrationTest;
import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
import org.apache.cassandra.utils.Shared;
@@ -61,13 +63,13 @@ class LeavingTest extends LeavingBaseTest
void retrieveMappingWithKeyspaceLeavingNode(VertxTestContext context,
ConfigurableCassandraTestContext cassandraTestContext) throws Exception
{
- BBHelperSingleLeavingNode.reset();
+ BBHelperLeavingNode.reset();
runLeavingTestScenario(context,
cassandraTestContext,
1,
- BBHelperSingleLeavingNode::install,
- BBHelperSingleLeavingNode.transientStateStart,
- BBHelperSingleLeavingNode.transientStateEnd,
+ (cl, nodeNum) -> BBHelperLeavingNode.install(cl, nodeNum, 5),
+ BBHelperLeavingNode.transientStateStart,
+ BBHelperLeavingNode.transientStateEnd,
generateExpectedRangeMappingSingleLeavingNode());
}
@@ -259,47 +261,6 @@ class LeavingTest extends LeavingBaseTest
};
}
- /**
- * ByteBuddy Helper for a single leaving node
- */
- public static class BBHelperSingleLeavingNode
- {
- static CountDownLatch transientStateStart = new CountDownLatch(1);
- static CountDownLatch transientStateEnd = new CountDownLatch(1);
-
- public static void install(ClassLoader cl, Integer nodeNumber)
- {
- // Test case involves 5 node cluster with 1 leaving node
- // We intercept the shutdown of the leaving node (5) to validate token ranges
- if (nodeNumber == 5)
- {
- TypePool typePool = TypePool.Default.of(cl);
- TypeDescription description = typePool.describe("org.apache.cassandra.service.StorageService")
- .resolve();
- new ByteBuddy().rebase(description, ClassFileLocator.ForClassLoader.of(cl))
- .method(named("unbootstrap"))
- .intercept(MethodDelegation.to(BBHelperSingleLeavingNode.class))
- // Defer class loading until all dependencies are loaded
- .make(TypeResolutionStrategy.Lazy.INSTANCE, typePool)
- .load(cl, ClassLoadingStrategy.Default.INJECTION);
- }
- }
-
- @SuppressWarnings("unused")
- public static void unbootstrap(@SuperCall Callable<?> orig) throws Exception
- {
- transientStateStart.countDown();
- awaitLatchOrTimeout(transientStateEnd, 2, TimeUnit.MINUTES, "transientStateEnd");
- orig.call();
- }
-
- public static void reset()
- {
- transientStateStart = new CountDownLatch(1);
- transientStateEnd = new CountDownLatch(1);
- }
- }
-
/**
* ByteBuddy helper for multiple leaving nodes
*/
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/MovingBaseTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/MovingBaseTest.java
index 8a8aae99..d08fa870 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/MovingBaseTest.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/MovingBaseTest.java
@@ -44,6 +44,7 @@ import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
import org.apache.cassandra.sidecar.common.server.cluster.locator.Partitioners;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
import org.apache.cassandra.testing.CassandraIntegrationTest;
import static org.assertj.core.api.Assertions.assertThat;
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/MovingTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/MovingTest.java
index 86dc4949..90a691c1 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/MovingTest.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/MovingTest.java
@@ -23,36 +23,24 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
import com.google.common.collect.Range;
import org.junit.jupiter.api.extension.ExtendWith;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
-import net.bytebuddy.ByteBuddy;
-import net.bytebuddy.description.type.TypeDescription;
-import net.bytebuddy.dynamic.ClassFileLocator;
-import net.bytebuddy.dynamic.TypeResolutionStrategy;
-import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
-import net.bytebuddy.implementation.MethodDelegation;
-import net.bytebuddy.implementation.bind.annotation.SuperCall;
-import net.bytebuddy.pool.TypePool;
import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
+import org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperMovingNode;
import org.apache.cassandra.testing.CassandraIntegrationTest;
import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
-import static net.bytebuddy.matcher.ElementMatchers.named;
-
/**
* Node movement scenarios integration tests for token range replica mapping endpoint with the in-jvm dtest framework.
*/
@ExtendWith(VertxExtension.class)
-class MovingTest extends MovingBaseTest
+public class MovingTest extends MovingBaseTest
{
@CassandraIntegrationTest(nodesPerDc = 5, network = true, buildCluster = false)
@@ -67,7 +55,7 @@ class MovingTest extends MovingBaseTest
1);
UpgradeableCluster cluster = cassandraTestContext.configureAndStartCluster(builder -> {
- builder.withInstanceInitializer(BBHelperMovingNode::install);
+ builder.withInstanceInitializer((cl, num) -> BBHelperMovingNode.install(cl, num, MOVING_NODE_IDX));
builder.withTokenSupplier(tokenSupplier);
});
@@ -128,45 +116,4 @@ class MovingTest extends MovingBaseTest
}
};
}
-
- /**
- * ByteBuddy Helper for a single moving node
- */
- public static class BBHelperMovingNode
- {
- static CountDownLatch transientStateStart = new CountDownLatch(1);
- static CountDownLatch transientStateEnd = new CountDownLatch(1);
-
- public static void install(ClassLoader cl, Integer nodeNumber)
- {
- // Moving the 5th node in the test case
- if (nodeNumber == MOVING_NODE_IDX)
- {
- TypePool typePool = TypePool.Default.of(cl);
- TypeDescription description = typePool.describe("org.apache.cassandra.service.RangeRelocator")
- .resolve();
- new ByteBuddy().rebase(description, ClassFileLocator.ForClassLoader.of(cl))
- .method(named("stream"))
- .intercept(MethodDelegation.to(BBHelperMovingNode.class))
- // Defer class loading until all dependencies are loaded
- .make(TypeResolutionStrategy.Lazy.INSTANCE, typePool)
- .load(cl, ClassLoadingStrategy.Default.INJECTION);
- }
- }
-
- @SuppressWarnings("unused")
- public static Future<?> stream(@SuperCall Callable<Future<?>> orig) throws Exception
- {
- Future<?> res = orig.call();
- transientStateStart.countDown();
- awaitLatchOrTimeout(transientStateEnd, 2, TimeUnit.MINUTES, "transientStateEnd");
- return res;
- }
-
- public static void reset()
- {
- transientStateStart = new CountDownLatch(1);
- transientStateEnd = new CountDownLatch(1);
- }
- }
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/ReplacementMultiDCTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/ReplacementMultiDCTest.java
index e0cf562e..9926666e 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/ReplacementMultiDCTest.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/ReplacementMultiDCTest.java
@@ -170,7 +170,7 @@ class ReplacementMultiDCTest extends ReplacementBaseTest
// We intercept the bootstrap of the replacement nodes to validate token ranges
if (nodeNumber > 10)
{
- BootstrapBBUtils.installSetBoostrapStateIntercepter(cl, BBHelperReplacementsMultiDC.class);
+ BootstrapBBUtils.installSetBoostrapStateInterceptor(cl, BBHelperReplacementsMultiDC.class);
}
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/ReplacementTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/ReplacementTest.java
index 4115ef8b..4d76f5b6 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/ReplacementTest.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/ReplacementTest.java
@@ -40,6 +40,7 @@ import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.IUpgradeableInstance;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.sidecar.testing.BootstrapBBUtils;
+import org.apache.cassandra.sidecar.testing.TestTokenSupplier;
import org.apache.cassandra.testing.CassandraIntegrationTest;
import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
@@ -150,7 +151,7 @@ class ReplacementTest extends ReplacementBaseTest
// We intercept the bootstrap of the replacement (6th) node to validate token ranges
if (nodeNumber == 6)
{
- BootstrapBBUtils.installSetBoostrapStateIntercepter(cl, BBHelperReplacementsNode.class);
+ BootstrapBBUtils.installSetBoostrapStateInterceptor(cl, BBHelperReplacementsNode.class);
}
}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/testing/BootstrapBBUtils.java b/server/src/test/integration/org/apache/cassandra/sidecar/testing/BootstrapBBUtils.java
index a3420e2f..fd6ec691 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/testing/BootstrapBBUtils.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/testing/BootstrapBBUtils.java
@@ -38,7 +38,7 @@ public class BootstrapBBUtils
* Note that the test class _must_ define the `setBootstrapState` method in order for the installed intercepter to be effective.
* See {@code ReplacementTest.BBHelperReplacementsNode} for example
*/
- public static void installSetBoostrapStateIntercepter(ClassLoader cl, Class<?> interceptor)
+ public static void installSetBoostrapStateInterceptor(ClassLoader cl, Class<?> interceptor)
{
TypePool typePool = TypePool.Default.of(cl);
TypeDescription description = typePool.describe("org.apache.cassandra.db.SystemKeyspace")
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/TestTokenSupplier.java b/server/src/test/integration/org/apache/cassandra/sidecar/testing/TestTokenSupplier.java
similarity index 86%
rename from server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/TestTokenSupplier.java
rename to server/src/test/integration/org/apache/cassandra/sidecar/testing/TestTokenSupplier.java
index 4aa04187..f0dee3a8 100644
--- a/server/src/test/integration/org/apache/cassandra/sidecar/routes/tokenrange/TestTokenSupplier.java
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/testing/TestTokenSupplier.java
@@ -16,10 +16,11 @@
* limitations under the License.
*/
-package org.apache.cassandra.sidecar.routes.tokenrange;
+package org.apache.cassandra.sidecar.testing;
import java.math.BigInteger;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import org.apache.cassandra.distributed.api.TokenSupplier;
@@ -39,10 +40,10 @@ public class TestTokenSupplier
* @param numTokensPerNode no. tokens allocated to each node (this is always 1 if there are no vnodes)
* @return The token supplier that vends the tokens
*/
- static TokenSupplier evenlyDistributedTokens(int numNodesPerDC, int newNodesPerDC, int numDcs, int numTokensPerNode)
+ public static TokenSupplier evenlyDistributedTokens(int numNodesPerDC, int newNodesPerDC, int numDcs, int numTokensPerNode)
{
// Use token count using initial node count to first assign tokens to nodes
- long totalTokens = (long) (numNodesPerDC) * numDcs * numTokensPerNode;
+ long totalTokens = (long) numNodesPerDC * numDcs * numTokensPerNode;
// Similar to Cassandra TokenSupplier, the increment is doubled to account for all tokens from MIN - MAX.
// For multi-DC, since neighboring nodes from different DCs have consecutive tokens, the increment is
// broadened by a factor of numDcs.
@@ -55,7 +56,7 @@ public class TestTokenSupplier
// Initial value of the first new node
- BigInteger value = new BigInteger(tokens[0 + (numDcs - 1)].get(0));
+ BigInteger value = new BigInteger(tokens[numDcs - 1].get(0));
BigInteger subIncrement = increment.divide(BigInteger.valueOf(2));
int nodeId = (int) totalTokens + 1;
@@ -77,6 +78,16 @@ public class TestTokenSupplier
return (nodeIdx) -> tokens[nodeIdx - 1];
}
+ /**
+ * Statically allocated tokens, one per node.
+ * @param tokens array of tokens to use
+ * @return TokenSupplier to vend tokens
+ */
+ public static TokenSupplier staticTokens(long... tokens)
+ {
+ return nodeIndex -> Collections.singletonList(String.valueOf(tokens[nodeIndex - 1]));
+ }
+
private static List<String>[] allocateExistingNodeTokens(int numNodesPerDC,
int newNodesPerDC,
int numDcs,
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperJoiningNode.java b/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperJoiningNode.java
new file mode 100644
index 00000000..eccfc941
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperJoiningNode.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.testing.bytebuddy;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.sidecar.testing.BootstrapBBUtils;
+
+import static org.apache.cassandra.sidecar.testing.IntegrationTestBase.awaitLatchOrTimeout;
+
+/**
+ * ByteBuddy Helper for a single node joining
+ * Note that the helper cannot be used by multiple tests simultaneously in the same JVM
+ */
+public class BBHelperJoiningNode
+{
+ public static CountDownLatch transientStateStart = new CountDownLatch(1);
+ public static CountDownLatch transientStateEnd = new CountDownLatch(1);
+
+ public static void install(ClassLoader cl, int nodeNumber, int joiningNodeIndex)
+ {
+ if (nodeNumber == joiningNodeIndex)
+ {
+ BootstrapBBUtils.installSetBoostrapStateInterceptor(cl, BBHelperJoiningNode.class);
+ }
+ }
+
+ public static void setBootstrapState(SystemKeyspace.BootstrapState state, @SuperCall Callable<Void> orig) throws Exception
+ {
+ if (state == SystemKeyspace.BootstrapState.COMPLETED)
+ {
+ // trigger bootstrap start and wait until bootstrap is ready from test
+ transientStateStart.countDown();
+ awaitLatchOrTimeout(transientStateEnd, 2, TimeUnit.MINUTES, "transientStateEnd");
+ }
+ orig.call();
+ }
+
+ public static void reset()
+ {
+ transientStateStart = new CountDownLatch(1);
+ transientStateEnd = new CountDownLatch(1);
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperLeavingNode.java b/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperLeavingNode.java
new file mode 100644
index 00000000..12322c05
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperLeavingNode.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.testing.bytebuddy;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.dynamic.ClassFileLocator;
+import net.bytebuddy.dynamic.TypeResolutionStrategy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import net.bytebuddy.pool.TypePool;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.cassandra.sidecar.testing.IntegrationTestBase.awaitLatchOrTimeout;
+
+/**
+ * ByteBuddy Helper for a single node leaving
+ * Note that the helper cannot be used by multiple tests simultaneously in the same JVM
+ */
+public class BBHelperLeavingNode
+{
+ public static CountDownLatch transientStateStart = new CountDownLatch(1);
+ public static CountDownLatch transientStateEnd = new CountDownLatch(1);
+
+ public static void install(ClassLoader cl, int nodeNum, int leavingNodeNum)
+ {
+ if (nodeNum == leavingNodeNum)
+ {
+ TypePool typePool = TypePool.Default.of(cl);
+ TypeDescription description = typePool.describe("org.apache.cassandra.service.StorageService")
+ .resolve();
+ new ByteBuddy().rebase(description, ClassFileLocator.ForClassLoader.of(cl))
+ .method(named("unbootstrap"))
+ .intercept(MethodDelegation.to(BBHelperLeavingNode.class))
+ // Defer class loading until all dependencies are loaded
+ .make(TypeResolutionStrategy.Lazy.INSTANCE, typePool)
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ }
+ }
+
+ @SuppressWarnings("unused")
+ public static void unbootstrap(@SuperCall Callable<?> orig) throws Exception
+ {
+ transientStateStart.countDown();
+ awaitLatchOrTimeout(transientStateEnd, 2, TimeUnit.MINUTES, "transientStateEnd");
+ orig.call();
+ }
+
+ public static void reset()
+ {
+ transientStateStart = new CountDownLatch(1);
+ transientStateEnd = new CountDownLatch(1);
+ }
+}
diff --git a/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperMovingNode.java b/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperMovingNode.java
new file mode 100644
index 00000000..8e19e0c9
--- /dev/null
+++ b/server/src/test/integration/org/apache/cassandra/sidecar/testing/bytebuddy/BBHelperMovingNode.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.testing.bytebuddy;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.dynamic.ClassFileLocator;
+import net.bytebuddy.dynamic.TypeResolutionStrategy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import net.bytebuddy.pool.TypePool;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.cassandra.sidecar.testing.IntegrationTestBase.awaitLatchOrTimeout;
+
+/**
+ * ByteBuddy Helper for a single moving node
+ * Note that the helper cannot be used by multiple tests simultaneously in the same JVM
+ */
+public class BBHelperMovingNode
+{
+ public static CountDownLatch transientStateStart = new CountDownLatch(1);
+ public static CountDownLatch transientStateEnd = new CountDownLatch(1);
+
+ public static void install(ClassLoader cl, int nodeNumber, int movingNodeIndex)
+ {
+ if (nodeNumber == movingNodeIndex)
+ {
+ TypePool typePool = TypePool.Default.of(cl);
+ TypeDescription description = typePool.describe("org.apache.cassandra.service.RangeRelocator")
+ .resolve();
+ new ByteBuddy().rebase(description, ClassFileLocator.ForClassLoader.of(cl))
+ .method(named("stream"))
+ .intercept(MethodDelegation.to(BBHelperMovingNode.class))
+ // Defer class loading until all dependencies are loaded
+ .make(TypeResolutionStrategy.Lazy.INSTANCE, typePool)
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ }
+ }
+
+ @SuppressWarnings("unused")
+ public static Future<?> stream(@SuperCall Callable<Future<?>> orig) throws Exception
+ {
+ Future<?> res = orig.call();
+ transientStateStart.countDown();
+ awaitLatchOrTimeout(transientStateEnd, 2, TimeUnit.MINUTES, "transientStateEnd");
+ return res;
+ }
+
+ public static void reset()
+ {
+ transientStateStart = new CountDownLatch(1);
+ transientStateEnd = new CountDownLatch(1);
+ }
+}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/db/RestoreSliceTest.java b/server/src/test/java/org/apache/cassandra/sidecar/db/RestoreSliceTest.java
index 3f2f3141..9ef82178 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/db/RestoreSliceTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/db/RestoreSliceTest.java
@@ -19,17 +19,14 @@
package org.apache.cassandra.sidecar.db;
import java.math.BigInteger;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
import java.util.UUID;
import org.junit.jupiter.api.Test;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.utils.UUIDs;
-import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
-import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse.ReplicaInfo;
+import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -84,51 +81,46 @@ class RestoreSliceTest
}
@Test
- void testNoSplit()
+ void testCreateFromPayload()
{
- RestoreSlice slice = createTestingSlice(UUIDs.timeBased(), "slice-id", 0L, 10L);
- List<RestoreSlice> result = slice.splitMaybe(null);
- assertThat(result).hasSize(1);
- assertThat(result.get(0)).isSameAs(slice);
-
- TokenRangeReplicasResponse topology = mock(TokenRangeReplicasResponse.class);
- when(topology.writeReplicas()).thenReturn(Collections.singletonList(new ReplicaInfo("-10", "10", null)));
- // range in topology fully encloses the slice
- result = slice.splitMaybe(topology);
- assertThat(result).hasSize(1);
- assertThat(result.get(0)).isSameAs(slice);
+ CreateSliceRequestPayload payload = new CreateSliceRequestPayload("slice-id", 0, "bucket", "key", "checksum",
+ BigInteger.ONE, // first token
+ BigInteger.TEN, // end token
+ 123L, // uncompressed size
+ 100L); // compressed size
+ RestoreSlice slice = RestoreSlice.builder().createSliceRequestPayload(payload).build();
+ assertThat(slice.sliceId()).isEqualTo("slice-id");
+ assertThat(slice.bucketId()).isEqualTo((short) 0);
+ assertThat(slice.bucket()).isEqualTo("bucket");
+ assertThat(slice.key()).isEqualTo("key");
+ assertThat(slice.checksum()).isEqualTo("checksum");
+ assertThat(slice.startToken())
+ .describedAs("Start token is exclusive end in the range")
+ .isEqualTo(BigInteger.ZERO);
+ assertThat(slice.endToken()).isEqualTo(BigInteger.TEN);
+ assertThat(slice.uncompressedSize()).isEqualTo(123L);
+ assertThat(slice.compressedSize()).isEqualTo(100L);
}
@Test
- void testSplitNoOverlap()
+ void testNoSplit()
{
RestoreSlice slice = createTestingSlice(UUIDs.timeBased(), "slice-id", 0L, 10L);
- TokenRangeReplicasResponse topology = mock(TokenRangeReplicasResponse.class);
- when(topology.writeReplicas()).thenReturn(Collections.singletonList(new ReplicaInfo("100", "110", null)));
- // (0, 10] does not overlap with (100, 110]
- assertThatThrownBy(() -> slice.splitMaybe(topology))
- .isExactlyInstanceOf(IllegalStateException.class)
- .hasMessage("Token range of the slice is not found in the write replicas. slice range: (0, 10]");
+ RestoreSlice result = slice.trimMaybe(new TokenRange(-10L, 10L));
+ assertThat(result)
+ .describedAs("No trim is done when fully enclosed by the local token range")
+ .isSameAs(slice);
}
@Test
- void testSplitIntoMultiple()
+ void testSplitNoIntersection()
{
- RestoreSlice slice = createTestingSlice(UUIDs.timeBased(), "slice-id", 2L, 25L);
- TokenRangeReplicasResponse topology = mock(TokenRangeReplicasResponse.class);
- ReplicaInfo r1 = new ReplicaInfo("-10", "10", null);
- ReplicaInfo r2 = new ReplicaInfo("10", "20", null);
- ReplicaInfo r3 = new ReplicaInfo("20", "30", null);
- ReplicaInfo r4 = new ReplicaInfo("30", "40", null);
- when(topology.writeReplicas()).thenReturn(Arrays.asList(r1, r2, r3, r4));
- List<RestoreSlice> result = slice.splitMaybe(topology);
- assertThat(result).hasSize(3);
- assertThat(result.get(0).startToken()).isEqualTo(BigInteger.valueOf(2L));
- assertThat(result.get(0).endToken()).isEqualTo(BigInteger.valueOf(10L));
- assertThat(result.get(1).startToken()).isEqualTo(BigInteger.valueOf(10L));
- assertThat(result.get(1).endToken()).isEqualTo(BigInteger.valueOf(20L));
- assertThat(result.get(2).startToken()).isEqualTo(BigInteger.valueOf(20L));
- assertThat(result.get(2).endToken()).isEqualTo(BigInteger.valueOf(25L));
+ RestoreSlice slice = createTestingSlice(UUIDs.timeBased(), "slice-id", 0L, 10L);
+ // (0, 10] does not intersect with (100, 110]
+ assertThatThrownBy(() -> slice.trimMaybe(new TokenRange(100L, 110L)))
+ .isExactlyInstanceOf(IllegalStateException.class)
+ .hasMessage("Token range of the slice does not intersect with the local token range. " +
+ "slice_range: TokenRange(0, 10], local_range: TokenRange(100, 110]");
}
public static RestoreSlice createTestingSlice(RestoreJob restoreJob, String sliceId, long startToken, long endToken)
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/db/SidecarSchemaTest.java b/server/src/test/java/org/apache/cassandra/sidecar/db/SidecarSchemaTest.java
index 8a88fa05..c17a46c5 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/db/SidecarSchemaTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/db/SidecarSchemaTest.java
@@ -168,10 +168,10 @@ public class SidecarSchemaTest
"SELECT job_id, bucket_id, slice_id, bucket, key, checksum, start_token, end_token, compressed_size, " +
"uncompressed_size FROM sidecar_internal.restore_slice_v3 " +
- "WHERE job_id = ? AND bucket_id = ? AND end_token >= ? AND start_token <= ? ALLOW FILTERING",
+ "WHERE job_id = ? AND bucket_id = ? AND end_token > ? AND start_token < ? ALLOW FILTERING",
- "INSERT INTO sidecar_internal.restore_range_v1 ( job_id, bucket_id, start_token, end_token, " +
- "slice_id, slice_bucket, slice_key, status_by_replica) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ "UPDATE sidecar_internal.restore_range_v1 SET slice_id = ?, slice_bucket = ?, slice_key = ?, status_by_replica = status_by_replica + ? " +
+ "WHERE job_id = ? AND bucket_id = ? AND start_token = ? AND end_token = ?",
"SELECT job_id, bucket_id, slice_id, slice_bucket, slice_key, start_token, end_token, status_by_replica " +
"FROM sidecar_internal.restore_range_v1 WHERE job_id = ? AND bucket_id = ? ALLOW FILTERING",
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyLevelCheckerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyCheckerTest.java
similarity index 64%
rename from server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyLevelCheckerTest.java
rename to server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyCheckerTest.java
index 90a5eeb4..abaf499b 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyLevelCheckerTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyCheckerTest.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.sidecar.restore;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -25,6 +26,10 @@ import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Range;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.jupiter.api.Test;
import org.apache.cassandra.sidecar.cluster.ConsistencyVerifier;
@@ -34,23 +39,26 @@ import org.apache.cassandra.sidecar.common.data.ConsistencyLevel;
import org.apache.cassandra.sidecar.common.data.ConsistencyVerificationResult;
import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse.ReplicaInfo;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.Token;
import org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus;
import org.apache.cassandra.sidecar.db.RestoreRange;
+import org.assertj.core.api.MapAssert;
-import static org.apache.cassandra.sidecar.restore.RestoreJobConsistencyLevelChecker.concludeOneRangeUnsafe;
-import static org.apache.cassandra.sidecar.restore.RestoreJobConsistencyLevelChecker.replicaSetForRangeUnsafe;
+import static org.apache.cassandra.sidecar.restore.RestoreJobConsistencyChecker.concludeOneRangeUnsafe;
+import static org.apache.cassandra.sidecar.restore.RestoreJobConsistencyChecker.populateStatusByReplica;
+import static org.apache.cassandra.sidecar.restore.RestoreJobConsistencyChecker.replicaSetForRangeUnsafe;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-class RestoreJobConsistencyLevelCheckerTest
+class RestoreJobConsistencyCheckerTest
{
@Test
void testReplicaSetForRangeNotFound()
{
RestoreRange range = RestoreRangeTest.createTestRange(1, 10);
TokenRangeReplicasResponse topology = mock(TokenRangeReplicasResponse.class);
- ReplicaInfo r1 = new ReplicaInfo("100", "110", null);
+ ReplicaInfo r1 = new ReplicaInfo("100", "110", ImmutableMap.of("dc1", ImmutableList.of("i1", "i2")));
when(topology.writeReplicas()).thenReturn(Collections.singletonList(r1));
InstanceSetByDc result = replicaSetForRangeUnsafe(range, topology);
assertThat(result).isNull();
@@ -65,7 +73,7 @@ class RestoreJobConsistencyLevelCheckerTest
{
RestoreRange range = RestoreRangeTest.createTestRange(1, 10);
TokenRangeReplicasResponse topology = mock(TokenRangeReplicasResponse.class);
- ReplicaInfo r1 = new ReplicaInfo("5", "15", null);
+ ReplicaInfo r1 = new ReplicaInfo("5", "15", ImmutableMap.of("dc1", ImmutableList.of("i1", "i2")));
when(topology.writeReplicas()).thenReturn(Collections.singletonList(r1));
InstanceSetByDc result = replicaSetForRangeUnsafe(range, topology);
assertThat(result)
@@ -168,6 +176,68 @@ class RestoreJobConsistencyLevelCheckerTest
.isEqualTo(ConsistencyVerificationResult.FAILED);
}
+ @Test
+ void testConcludeOneRangeWithDiscarded()
+ {
+ RestoreRange range = RestoreRangeTest.createTestRange(1, 10);
+ TokenRangeReplicasResponse topology = mock(TokenRangeReplicasResponse.class);
+ ReplicaInfo r1 = new ReplicaInfo("0", "15", replicaByDc(1, 3));
+ when(topology.writeReplicas()).thenReturn(Collections.singletonList(r1));
+
+ ConsistencyVerifier localQuorumVerifier = ConsistencyVerifiers.forConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM, "dc-1");
+
+ // has quorum replicas of desired status
+ Map<String, RestoreRangeStatus> replicaStatus = new HashMap<>();
+ replicaStatus.put("i-1", RestoreRangeStatus.STAGED);
+ replicaStatus.put("i-2", RestoreRangeStatus.CREATED);
+ replicaStatus.put("i-3", RestoreRangeStatus.FAILED);
+ replicaStatus.put("i-4", RestoreRangeStatus.DISCARDED); // i-4 worked on the range, but discards it since it loses the ownership
+ range = range.unbuild().replicaStatus(replicaStatus).build();
+ assertThat(concludeOneRangeUnsafe(topology, localQuorumVerifier, RestoreRangeStatus.STAGED, range))
+ .isEqualTo(ConsistencyVerificationResult.PENDING);
+
+ replicaStatus.put("i-1", RestoreRangeStatus.STAGED);
+ replicaStatus.put("i-2", RestoreRangeStatus.STAGED);
+ replicaStatus.put("i-3", RestoreRangeStatus.FAILED);
+ replicaStatus.put("i-4", RestoreRangeStatus.DISCARDED); // i-4 worked on the range, but discards it since it loses the ownership
+ range = range.unbuild().replicaStatus(replicaStatus).build();
+ assertThat(concludeOneRangeUnsafe(topology, localQuorumVerifier, RestoreRangeStatus.STAGED, range))
+ .isEqualTo(ConsistencyVerificationResult.SATISFIED);
+ }
+
+ @Test
+ void testPopulateStatusByReplica()
+ {
+ List<RestoreRange> restoreRanges = Arrays.asList(r(1, 10, ImmutableMap.of("i1", RestoreRangeStatus.STAGED,
+ "i2", RestoreRangeStatus.CREATED)),
+ r(5, 15, ImmutableMap.of("i3", RestoreRangeStatus.STAGED)),
+ r(15, 20, ImmutableMap.of("i3", RestoreRangeStatus.STAGED,
+ "i4", RestoreRangeStatus.CREATED)));
+ List<Range<Token>> expectedKeys =
+ Arrays.asList(Range.openClosed(Token.from(1), Token.from(5)),
+ Range.openClosed(Token.from(5), Token.from(10)), // overlapping range of (1, 10] and (5, 15]
+ Range.openClosed(Token.from(10), Token.from(15)),
+ Range.openClosed(Token.from(15), Token.from(20)));
+ List<Map<String, RestoreRangeStatus>> expectedStatus =
+ Arrays.asList(ImmutableMap.of("i1", RestoreRangeStatus.STAGED,
+ "i2", RestoreRangeStatus.CREATED),
+ ImmutableMap.of("i1", RestoreRangeStatus.STAGED,
+ "i2", RestoreRangeStatus.CREATED,
+ "i3", RestoreRangeStatus.STAGED), // merged from range (5, 15]
+ ImmutableMap.of("i3", RestoreRangeStatus.STAGED),
+ ImmutableMap.of("i3", RestoreRangeStatus.STAGED,
+ "i4", RestoreRangeStatus.CREATED));
+ Map<Range<Token>, Pair<Map<String, RestoreRangeStatus>, RestoreRange>> result = populateStatusByReplica(restoreRanges);
+ MapAssert<Range<Token>, ?> assertion = assertThat(result).hasSize(4);
+ for (int i = 0; i < 4; i++)
+ {
+ Range<Token> expectedKey = expectedKeys.get(i);
+ assertion.containsKey(expectedKey);
+ Map<String, RestoreRangeStatus> status = result.get(expectedKey).getLeft();
+ assertThat(status).isEqualTo(expectedStatus.get(i));
+ }
+ }
+
private Map<String, List<String>> replicaByDc(int dcCount, int replicasPerDc)
{
Map<String, List<String>> result = new HashMap<>(dcCount);
@@ -178,4 +248,9 @@ class RestoreJobConsistencyLevelCheckerTest
}
return result;
}
+
+ private static RestoreRange r(long start, long end, Map<String, RestoreRangeStatus> status)
+ {
+ return RestoreRangeTest.createTestRange(start, end).unbuild().replicaStatus(status).build();
+ }
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
index 427bd8b9..8050a753 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
@@ -36,10 +36,10 @@ import com.datastax.driver.core.LocalDate;
import com.datastax.driver.core.utils.UUIDs;
import io.vertx.core.Promise;
import org.apache.cassandra.sidecar.TestModule;
-import org.apache.cassandra.sidecar.cluster.locator.LocalTokenRangesProvider;
import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
import org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
import org.apache.cassandra.sidecar.config.RestoreJobConfiguration;
import org.apache.cassandra.sidecar.db.RestoreJob;
import org.apache.cassandra.sidecar.db.RestoreJobDatabaseAccessor;
@@ -74,7 +74,6 @@ class RestoreJobDiscovererTest
private final RestoreSliceDatabaseAccessor mockSliceAccessor = mock(RestoreSliceDatabaseAccessor.class);
private final RestoreRangeDatabaseAccessor mockRangeAccessor = mock(RestoreRangeDatabaseAccessor.class);
private final RestoreJobManagerGroup mockManagers = mock(RestoreJobManagerGroup.class);
- private final LocalTokenRangesProvider rangesProvider = mock(LocalTokenRangesProvider.class);
private final PeriodicTaskExecutor executor = mock(PeriodicTaskExecutor.class);
private final SidecarSchema sidecarSchema = mock(SidecarSchema.class);
private SidecarMetrics metrics;
@@ -93,9 +92,9 @@ class RestoreJobDiscovererTest
mockSliceAccessor,
mockRangeAccessor,
() -> mockManagers,
- rangesProvider,
null,
null,
+ mock(ExecutorPools.class),
metrics);
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobManagerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobManagerTest.java
index 1fc5e776..9bc00f79 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobManagerTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobManagerTest.java
@@ -19,10 +19,13 @@
package org.apache.cassandra.sidecar.restore;
import java.io.IOException;
+import java.math.BigInteger;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@@ -39,6 +42,7 @@ import io.vertx.core.Vertx;
import org.apache.cassandra.sidecar.ExecutorPoolsHelper;
import org.apache.cassandra.sidecar.TestModule;
import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
import org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
@@ -54,15 +58,18 @@ import org.apache.cassandra.sidecar.server.MainModule;
import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
class RestoreJobManagerTest
{
+ private static final int jobRecencyDays = 1;
private RestoreJobManager manager;
private Vertx vertx;
private ExecutorPools executorPools;
- private static final int jobRecencyDays = 1;
+ private RestoreProcessor processor;
@TempDir
private Path testDir;
@@ -73,7 +80,7 @@ class RestoreJobManagerTest
Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(new TestModule()));
vertx = injector.getInstance(Vertx.class);
executorPools = ExecutorPoolsHelper.createdSharedTestPool(vertx);
- RestoreProcessor processor = mock(RestoreProcessor.class);
+ processor = mock(RestoreProcessor.class);
InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
when(instanceMetadata.stagingDir()).thenReturn(testDir.toString());
@@ -148,12 +155,15 @@ class RestoreJobManagerTest
assertThat(manager.trySubmit(range, range.job()))
.isEqualTo(RestoreJobProgressTracker.Status.CREATED);
- assertThat(range.isCancelled()).isFalse();
+ Map<RestoreRange, ?> ranges = manager.progressTrackerUnsafe(range.job()).rangesForTesting();
+ assertThat(ranges.size()).isOne();
+ RestoreRange submittedRange = ranges.keySet().iterator().next();
+ assertThat(submittedRange.isCancelled()).isFalse();
- manager.removeJobInternal(range.jobId()); // it cancels the non-completed ranges
+ manager.removeJobInternal(submittedRange.jobId()); // it cancels the non-completed ranges
// removeJobInternal runs async. Wait for at most 2 seconds for the slice to be cancelled
- loopAssert(2, () -> assertThat(range.isCancelled()).isTrue());
+ loopAssert(2, () -> assertThat(submittedRange.isCancelled()).isTrue());
}
@Test
@@ -245,6 +255,52 @@ class RestoreJobManagerTest
});
}
+ @Test
+ void testDiscardOverlappingRanges() throws Exception
+ {
+ // set up the mock for discardAndRemove; invoke discard on the input restore range
+ doAnswer(invocation -> {
+ RestoreRange input = invocation.getArgument(0, RestoreRange.class);
+ input.discard();
+ return null;
+ })
+ .when(processor).discardAndRemove(any(RestoreRange.class));
+ RestoreRange template = getTestRange();
+ RestoreJob job = template.job();
+ RestoreRange rangeToDiscard = template.unbuild()
+ .sliceId("rangeToDiscard")
+ .startToken(BigInteger.valueOf(0))
+ .endToken(BigInteger.valueOf(10))
+ .build();
+ RestoreRange rangeToKeep = template.unbuild()
+ .sliceId("rangeToKeep")
+ .startToken(BigInteger.valueOf(100))
+ .endToken(BigInteger.valueOf(110))
+ .build();
+ assertThat(manager.progressTrackerUnsafe(job).rangesForTesting())
+ .describedAs("No range is submitted yet")
+ .hasSize(0);
+ assertThat(manager.trySubmit(rangeToDiscard, job))
+ .isEqualTo(RestoreJobProgressTracker.Status.CREATED);
+ assertThat(manager.trySubmit(rangeToKeep, job))
+ .isEqualTo(RestoreJobProgressTracker.Status.CREATED);
+
+ assertThat(manager.progressTrackerUnsafe(job).rangesForTesting())
+ .describedAs("There are two ranges submitted")
+ .hasSize(2);
+ // (0, 10] overlaps with (-10, 50]; but (100, 110] does not
+ Set<RestoreRange> rangesDiscarded = manager.discardOverlappingRanges(job, Collections.singleton(new TokenRange(-10, 50)));
+ assertThat(rangesDiscarded).hasSize(1);
+ RestoreRange rangeDiscarded = rangesDiscarded.iterator().next();
+ assertThat(rangeDiscarded.isDiscarded()).isTrue();
+ assertThat(rangeDiscarded.sliceId()).isEqualTo("rangeToDiscard");
+ assertThat(manager.progressTrackerUnsafe(job).rangesForTesting())
+ .describedAs("One of the two ranges should be discarded")
+ .hasSize(1);
+ RestoreRange rangeKept = manager.progressTrackerUnsafe(job).rangesForTesting().keySet().iterator().next();
+ assertThat(rangeKept.sliceId()).isEqualTo("rangeToKeep");
+ }
+
private RestoreRange getTestRange()
{
return getTestRange(RestoreJobTest.createNewTestingJob(UUIDs.timeBased()));
@@ -261,6 +317,8 @@ class RestoreJobManagerTest
.bucketId((short) 0)
.storageKey("storageKey")
.storageBucket("storageBucket")
+ .startToken(BigInteger.ONE)
+ .endToken(BigInteger.TEN)
.build();
RestoreJobProgressTracker tracker = manager.progressTrackerUnsafe(job);
return RestoreRange.builderFromSlice(slice)
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreProcessorTest.java b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreProcessorTest.java
index a73f2a40..bf1a34cd 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreProcessorTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreProcessorTest.java
@@ -28,15 +28,12 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import com.datastax.driver.core.utils.UUIDs;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.util.Modules;
import io.vertx.core.Promise;
import org.apache.cassandra.sidecar.TestModule;
-import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
-import org.apache.cassandra.sidecar.db.RestoreJob;
import org.apache.cassandra.sidecar.db.RestoreRange;
import org.apache.cassandra.sidecar.db.schema.SidecarSchema;
import org.apache.cassandra.sidecar.exceptions.RestoreJobFatalException;
@@ -46,8 +43,8 @@ import org.apache.cassandra.sidecar.metrics.instance.InstanceRestoreMetrics;
import org.apache.cassandra.sidecar.server.MainModule;
import org.apache.cassandra.sidecar.tasks.PeriodicTaskExecutor;
import org.apache.cassandra.sidecar.tasks.ScheduleDecision;
-import org.mockito.Mockito;
+import static org.apache.cassandra.sidecar.restore.RestoreRangeTask.failOnCancelled;
import static org.apache.cassandra.sidecar.utils.TestMetricUtils.registry;
import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
import static org.assertj.core.api.Assertions.assertThat;
@@ -92,14 +89,16 @@ class RestoreProcessorTest
int concurrency = TestModule.RESTORE_MAX_CONCURRENCY;
periodicTaskExecutor.schedule(processor);
- assertThat(processor.activeSlices()).isZero();
+ assertThat(processor.activeRanges()).isZero();
CountDownLatch latch = new CountDownLatch(1);
int total = concurrency * 3;
for (int i = 0; i < total; i++)
{
- processor.submit(mockSlowRestoreRange(latch));
+ RestoreRange range = mockSlowRestoreRange(latch);
+ range.completeImportPhase(); // complete prematurely for testing purpose
+ processor.submit(range);
}
InstanceRestoreMetrics instanceRestoreMetrics = instanceMetrics().restore();
@@ -114,7 +113,7 @@ class RestoreProcessorTest
assertThat(instanceRestoreMetrics.pendingSliceCount.metric.getValue())
.isLessThanOrEqualTo(total - concurrency);
- assertThat(processor.activeSlices()).isEqualTo(concurrency);
+ assertThat(processor.activeRanges()).isEqualTo(concurrency);
});
// slices start to succeed
@@ -122,7 +121,7 @@ class RestoreProcessorTest
// it never grows beyond `concurrency`
loopAssert(3, () -> {
- assertThat(processor.activeSlices())
+ assertThat(processor.activeRanges())
.describedAs("Active slice count should be in the range of (0, concurrency]")
.isLessThanOrEqualTo(concurrency)
.isPositive();
@@ -131,7 +130,7 @@ class RestoreProcessorTest
// the active slices should be back to 0
// and the pending slices should be back to 0
loopAssert(3, () -> {
- assertThat(processor.activeSlices()).isZero();
+ assertThat(processor.activeRanges()).isZero();
assertThat(instanceRestoreMetrics.sliceImportQueueLength.metric.getValue()).isZero();
assertThat(instanceRestoreMetrics.pendingSliceCount.metric.getValue()).isZero();
});
@@ -150,19 +149,19 @@ class RestoreProcessorTest
when(sidecarSchema.isInitialized()).thenReturn(false);
assertThat(processor.scheduleDecision()).isEqualTo(ScheduleDecision.SKIP);
- assertThat(processor.activeSlices()).isZero();
+ assertThat(processor.activeRanges()).isZero();
CountDownLatch latch = new CountDownLatch(1);
processor.submit(mockSlowRestoreRange(latch));
- assertThat(processor.activeSlices())
+ assertThat(processor.activeRanges())
.describedAs("No slice should be active because executions are skipped")
.isZero();
// Make slice completable. But since all executions are skipped, the active slice should remain as 1
latch.countDown();
loopAssert(3, () -> {
- assertThat(processor.pendingStartSlices()).isOne();
- assertThat(processor.activeSlices()).isZero();
+ assertThat(processor.pendingStartRanges()).isOne();
+ assertThat(processor.activeRanges()).isZero();
});
}
@@ -220,7 +219,41 @@ class RestoreProcessorTest
assertThatThrownBy(() -> range.trackerUnsafe().trySubmit(range))
.isExactlyInstanceOf(RestoreJobFatalException.class)
.hasMessageContaining("Restore range is cancelled.");
+ }
+
+ @Test
+ void testDiscardPendingRange()
+ {
+ RestoreRange range = RestoreRangeTest.createTestRange();
+ processor.submit(range);
+ assertThat(processor.pendingStartRanges()).isOne();
+ processor.discardAndRemove(range);
+ assertThat(processor.pendingStartRanges()).isZero();
+ assertThat(range.isDiscarded()).isTrue();
+ }
+ @Test
+ void testDiscardStartedRange()
+ {
+ CountDownLatch readyToFinish = new CountDownLatch(1);
+ RestoreRange range = mockSlowRestoreRange(readyToFinish);
+ processor.submit(range);
+ processor.execute(Promise.promise());
+ assertThat(processor.activeRanges()).isOne();
+ assertThat(processor.activeTasks()).isOne();
+ processor.discardAndRemove(range);
+ assertThat(processor.activeRanges())
+ .describedAs("The range is being processed already, we wait for it to fail " +
+ "and discard in org.apache.cassandra.sidecar.restore.RestoreProcessor.taskFailureHandler")
+ .isOne();
+ assertThat(range.isDiscarded())
+ .describedAs("The range should be marked as discarded")
+ .isTrue();
+ readyToFinish.countDown();
+ loopAssert(3, () -> {
+ assertThat(processor.activeRanges()).isZero();
+ assertThat(processor.activeTasks()).isZero();
+ });
}
private InstanceMetrics instanceMetrics()
@@ -245,7 +278,8 @@ class RestoreProcessorTest
public void handle(Promise<RestoreRange> promise)
{
Uninterruptibles.awaitUninterruptibly(latch);
- promise.complete(range);
+ failOnCancelled(range, range)
+ .onComplete(promise);
}
@Override
@@ -265,17 +299,9 @@ class RestoreProcessorTest
private RestoreRange mockRestoreRange()
{
- RestoreRange range = mock(RestoreRange.class, Mockito.RETURNS_DEEP_STUBS);
- when(range.jobId()).thenReturn(UUIDs.timeBased());
- when(range.canProduceTask()).thenReturn(true);
- when(range.owner().id()).thenReturn(1);
- when(range.sliceKey()).thenReturn("SliceKey");
+ RestoreRange mockRange = RestoreRangeTest.createTestRange();
+ RestoreRange range = spy(mockRange);
when(range.owner().metrics()).thenReturn(instanceMetrics());
- RestoreJob job = RestoreJob.builder()
- .jobStatus(RestoreJobStatus.CREATED)
- .build();
- when(range.job()).thenReturn(job);
- when(range.hasImported()).thenReturn(true);
return range;
}
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreRangeTest.java b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreRangeTest.java
index 3f18ea0c..092fc1e0 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreRangeTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreRangeTest.java
@@ -19,6 +19,7 @@
package org.apache.cassandra.sidecar.restore;
import java.math.BigInteger;
+import java.net.InetSocketAddress;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collections;
@@ -40,6 +41,7 @@ import org.apache.cassandra.sidecar.db.RestoreRange;
import org.apache.cassandra.sidecar.db.RestoreSlice;
import org.apache.cassandra.sidecar.metrics.SidecarMetrics;
+import static org.apache.cassandra.sidecar.common.server.data.RestoreRangeStatus.DISCARDED;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
@@ -147,6 +149,17 @@ public class RestoreRangeTest
assertThat(restoreRangeJson.endToken()).isEqualTo(range.endToken());
}
+ @Test
+ void testDiscardRange()
+ {
+ RestoreRange range = createTestRange(true);
+ assertThat(range.isDiscarded()).isFalse();
+ range.discard();
+ assertThat(range.isDiscarded()).isTrue();
+ assertThat(range.isCancelled()).isTrue();
+ assertThat(range.statusByReplica()).containsEntry("127.0.0.1:12345", DISCARDED).hasSize(1);
+ }
+
private void assertFailedHandler(RestoreRange range, RestoreRangeHandler handler, String containsErrorMessage)
{
assertThat(handler).describedAs("It should create a Failed handler")
@@ -204,8 +217,9 @@ public class RestoreRangeTest
job = job.unbuild().consistencyLevel(ConsistencyLevel.QUORUM).build();
}
RestoreProcessor mockProcessor = mock(RestoreProcessor.class);
- InstanceMetadata mockInstance = mock(InstanceMetadata.class);
+ InstanceMetadata mockInstance = mock(InstanceMetadata.class, RETURNS_DEEP_STUBS);
when(mockInstance.id()).thenReturn(1);
+ when(mockInstance.delegate().localStorageBroadcastAddress()).thenReturn(new InetSocketAddress("127.0.0.1", 12345));
RestoreJobProgressTracker tracker = new RestoreJobProgressTracker(job, mockProcessor, mockInstance);
RestoreSlice slice = RestoreSlice.builder()
.jobId(job.jobId).keyspace("keyspace").table("table")
@@ -216,7 +230,7 @@ public class RestoreRangeTest
return RestoreRange.builderFromSlice(slice)
.stageDirectory(rootDir, "uploadId")
- .replicaStatus(Collections.singletonMap("replica1", RestoreRangeStatus.CREATED))
+ .replicaStatus(Collections.singletonMap("127.0.0.1:12345", RestoreRangeStatus.CREATED))
.restoreJobProgressTracker(tracker)
.ownerInstance(mockInstance)
.build();
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/restore/RingTopologyRefresherTest.java b/server/src/test/java/org/apache/cassandra/sidecar/restore/RingTopologyRefresherTest.java
index c0700307..982a065d 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/restore/RingTopologyRefresherTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/restore/RingTopologyRefresherTest.java
@@ -18,6 +18,11 @@
package org.apache.cassandra.sidecar.restore;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
import org.junit.jupiter.api.Test;
import com.datastax.driver.core.utils.UUIDs;
@@ -30,10 +35,14 @@ import org.apache.cassandra.sidecar.restore.RingTopologyRefresher.ReplicaByToken
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
class RingTopologyRefresherTest
{
- private final ReplicaByTokenRangePerKeyspace replicaByTokenRangePerKeyspace = new ReplicaByTokenRangePerKeyspace();
+ private final List<List<Object>> listenerValueCaptor = new ArrayList<>();
+ private final RingTopologyChangeListener testListener = (keyspace, oldTopology, newTopology) ->
+ listenerValueCaptor.add(Arrays.asList(keyspace, oldTopology, newTopology));
+ private final ReplicaByTokenRangePerKeyspace replicaByTokenRangePerKeyspace = new ReplicaByTokenRangePerKeyspace(testListener);
@Test
void testRegisterJobs()
@@ -67,6 +76,7 @@ class RingTopologyRefresherTest
assertThat(replicaByTokenRangePerKeyspace.jobsByKeyspaceUnsafe().get(job3.keyspaceName)).containsExactlyInAnyOrder(job3.jobId);
assertThat(replicaByTokenRangePerKeyspace.promisesUnsafe()).isEmpty();
assertThat(replicaByTokenRangePerKeyspace.mappingUnsafe()).isEmpty();
+ assertThat(listenerValueCaptor).isEmpty();
}
@Test
@@ -85,6 +95,7 @@ class RingTopologyRefresherTest
.isSameAs(future2);
assertThat(future1.isComplete()).isFalse();
assertThat(replicaByTokenRangePerKeyspace.mappingUnsafe()).isEmpty();
+ assertThat(listenerValueCaptor).isEmpty();
// load the topology should complete the futures
replicaByTokenRangePerKeyspace.load(ks -> mockTopology);
assertThat(future1.isComplete()).isTrue();
@@ -94,6 +105,8 @@ class RingTopologyRefresherTest
assertThat(replicaByTokenRangePerKeyspace.jobsByKeyspaceUnsafe()).containsOnlyKeys(job1.keyspaceName);
assertThat(replicaByTokenRangePerKeyspace.promisesUnsafe()).containsOnlyKeys(job1.keyspaceName);
assertThat(replicaByTokenRangePerKeyspace.mappingUnsafe()).containsEntry(job1.keyspaceName, mockTopology);
+ assertThat(listenerValueCaptor).hasSize(1);
+ assertListenerCapturedValues(listenerValueCaptor.get(0), job1.keyspaceName, null, mockTopology);
// now unregister the jobs. start with job1; because the job2 is still registered, the mapping remains
replicaByTokenRangePerKeyspace.unregister(job1);
@@ -141,10 +154,42 @@ class RingTopologyRefresherTest
Future<TokenRangeReplicasResponse> future = replicaByTokenRangePerKeyspace.futureOf(job);
assertThat(future.isComplete()).isFalse();
assertThat(replicaByTokenRangePerKeyspace.forRestoreJob(job)).isNull();
+ assertThat(listenerValueCaptor).isEmpty();
TokenRangeReplicasResponse mockTopology = mock(TokenRangeReplicasResponse.class);
replicaByTokenRangePerKeyspace.load(ks -> mockTopology);
assertThat(future.isComplete()).isTrue();
assertThat(future.result()).isSameAs(mockTopology);
assertThat(replicaByTokenRangePerKeyspace.forRestoreJob(job)).isSameAs(mockTopology);
+ assertThat(listenerValueCaptor).hasSize(1);
+ assertListenerCapturedValues(listenerValueCaptor.get(0), job.keyspaceName, null, mockTopology);
+ }
+
+ @Test
+ void testTopologyChanged()
+ {
+ assertThat(listenerValueCaptor).isEmpty();
+ RestoreJob job = RestoreJobTest.createNewTestingJob(UUIDs.timeBased());
+ replicaByTokenRangePerKeyspace.futureOf(job);
+ TokenRangeReplicasResponse mockTopologyEpoch1 = mock(TokenRangeReplicasResponse.class);
+ when(mockTopologyEpoch1.writeReplicas())
+ .thenReturn(Collections.singletonList(new TokenRangeReplicasResponse.ReplicaInfo("1", "10", null)));
+ replicaByTokenRangePerKeyspace.load(ks -> mockTopologyEpoch1);
+ assertThat(listenerValueCaptor).hasSize(1);
+ assertListenerCapturedValues(listenerValueCaptor.get(0), job.keyspaceName, null, mockTopologyEpoch1);
+ TokenRangeReplicasResponse mockTopologyEpoch2 = mock(TokenRangeReplicasResponse.class);
+ when(mockTopologyEpoch1.writeReplicas())
+ .thenReturn(Collections.singletonList(new TokenRangeReplicasResponse.ReplicaInfo("100", "110", null)));
+ replicaByTokenRangePerKeyspace.load(ks -> mockTopologyEpoch2);
+ assertThat(listenerValueCaptor).hasSize(2);
+ assertListenerCapturedValues(listenerValueCaptor.get(1), job.keyspaceName, mockTopologyEpoch1, mockTopologyEpoch2);
+ }
+
+ private void assertListenerCapturedValues(List<Object> values,
+ String keyspace, TokenRangeReplicasResponse oldValue, TokenRangeReplicasResponse newValue)
+ {
+ assertThat(values).hasSize(3);
+ assertThat(values.get(0)).isEqualTo(keyspace);
+ assertThat(values.get(1)).isSameAs(oldValue);
+ assertThat(values.get(2)).isSameAs(newValue);
}
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/routes/restore/BaseRestoreJobTests.java b/server/src/test/java/org/apache/cassandra/sidecar/routes/restore/BaseRestoreJobTests.java
index c1469461..5a109cc6 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/routes/restore/BaseRestoreJobTests.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/routes/restore/BaseRestoreJobTests.java
@@ -361,9 +361,9 @@ public abstract class BaseRestoreJobTests
{
Supplier<TokenRangeReplicasResponse> topologySupplier;
- public TestRingTopologyRefresher(InstanceMetadataFetcher metadataFetcher, SidecarConfiguration config)
+ public TestRingTopologyRefresher(InstanceMetadataFetcher metadataFetcher, SidecarConfiguration config, ExecutorPools executorPools)
{
- super(metadataFetcher, config);
+ super(metadataFetcher, config, executorPools);
}
@Override
@@ -416,9 +416,11 @@ public abstract class BaseRestoreJobTests
@Provides
@Singleton
- public RingTopologyRefresher ringTopologyRefresher(InstanceMetadataFetcher metadataFetcher, SidecarConfiguration configuration)
+ public RingTopologyRefresher ringTopologyRefresher(InstanceMetadataFetcher metadataFetcher,
+ SidecarConfiguration configuration,
+ ExecutorPools executorPools)
{
- return new TestRingTopologyRefresher(metadataFetcher, configuration);
+ return new TestRingTopologyRefresher(metadataFetcher, configuration, executorPools);
}
}
}
diff --git a/server/src/testFixtures/java/org/apache/cassandra/sidecar/config/yaml/TestServiceConfiguration.java b/server/src/testFixtures/java/org/apache/cassandra/sidecar/config/yaml/TestServiceConfiguration.java
index 08d27736..c8890c3b 100644
--- a/server/src/testFixtures/java/org/apache/cassandra/sidecar/config/yaml/TestServiceConfiguration.java
+++ b/server/src/testFixtures/java/org/apache/cassandra/sidecar/config/yaml/TestServiceConfiguration.java
@@ -23,7 +23,7 @@ import org.apache.cassandra.sidecar.config.ServiceConfiguration;
/**
* A ServiceConfiguration implementation created for test.
- * It binds to "127.0.0.1" and port 0 to find an available port at test runtime
+ * It binds to "0.0.0.0" and port 0 to find an available port at test runtime
*/
public class TestServiceConfiguration extends ServiceConfigurationImpl
{
@@ -35,7 +35,8 @@ public class TestServiceConfiguration extends ServiceConfigurationImpl
public static Builder builder()
{
return ServiceConfigurationImpl.builder()
- .host("127.0.0.1")
+ // bind to 0.0.0.0 to allow sidecar client to query all cassandra node, rather than only 127.0.0.1
+ .host("0.0.0.0")
.port(0); // let the test find an available port
}
}
diff --git a/spotbugs-exclude.xml b/spotbugs-exclude.xml
index 91e13ac5..2e53e520 100644
--- a/spotbugs-exclude.xml
+++ b/spotbugs-exclude.xml
@@ -120,4 +120,18 @@
<Class name="org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher" />
</Match>
+ <Match>
+ <Bug pattern="SF_SWITCH_FALLTHROUGH" />
+ <Class name="org.apache.cassandra.sidecar.restore.RestoreJobDiscoverer" /><!-- on break by design -->
+ </Match>
+
+ <Match>
+ <Bug pattern="MS_CANNOT_BE_FINAL" />
+ <Or>
+ <Class name="org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperMovingNode" /><!-- Test code -->
+ <Class name="org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperJoiningNode" /><!-- Test code -->
+ <Class name="org.apache.cassandra.sidecar.testing.bytebuddy.BBHelperLeavingNode" /><!-- Test code -->
+ </Or>
+ </Match>
+
</FindBugsFilter>
diff --git a/vertx-auth-mtls/build.gradle b/vertx-auth-mtls/build.gradle
index a9030918..b596f2a7 100644
--- a/vertx-auth-mtls/build.gradle
+++ b/vertx-auth-mtls/build.gradle
@@ -40,6 +40,9 @@ test {
html.setRequired(true)
html.getOutputLocation().set(destDir)
}
+ testLogging {
+ events "passed", "skipped", "failed"
+ }
}
configurations {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org