diff --git a/.circleci/config.yml b/.circleci/config.yml
index 81f64d53..34ce6e65 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -16,7 +16,7 @@ jobs:
- m2-{{ checksum "pom.xml" }}
- m2-
- - run: (echo "${google_application_credentials}" > /tmp/gcp.json) && mvn clean install -PsnapshotRepo,rpm,deb,cassandra5 -DoutputDirectory=/tmp/artifacts -Dcassandra4.version=4.1.10 -Dcassandra5.version=5.0.5
+ - run: (echo "${google_application_credentials}" > /tmp/gcp.json) && mvn clean install -PsnapshotRepo,rpm,deb,cassandra5 -DoutputDirectory=/tmp/artifacts -Dcassandra4.version=4.1.10 -Dcassandra5.version=5.0.6
- save_cache:
paths:
diff --git a/azure/pom.xml b/azure/pom.xml
index c29989cb..1a854a70 100644
--- a/azure/pom.xml
+++ b/azure/pom.xml
@@ -6,12 +6,12 @@
com.instaclustr
esop-parent
- 4.0.1
+ 4.0.2-SNAPSHOT
../pom.xml
esop-azure
- 4.0.1
+ 4.0.2-SNAPSHOT
esop-azure
Backup and restoration tooling for Cassandra for Azure
diff --git a/core/pom.xml b/core/pom.xml
index 718501b7..085355dc 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -6,12 +6,12 @@
com.instaclustr
esop-parent
- 4.0.1
+ 4.0.2-SNAPSHOT
../pom.xml
esop-core
- 4.0.1
+ 4.0.2-SNAPSHOT
esop-core
Core of backup and restoration tooling for Cassandra
@@ -25,6 +25,7 @@
2.20.0
3.1.6
2.19.2
+ 1.8.0
2.0.17
1.5.19
@@ -120,6 +121,12 @@
${awaitility.version}
+
+ org.lz4
+ lz4-java
+ ${lz4.version}
+
+
diff --git a/core/src/main/java/com/instaclustr/esop/cli/BackupApplication.java b/core/src/main/java/com/instaclustr/esop/cli/BackupApplication.java
index d2f91cfb..b2b0bf8d 100644
--- a/core/src/main/java/com/instaclustr/esop/cli/BackupApplication.java
+++ b/core/src/main/java/com/instaclustr/esop/cli/BackupApplication.java
@@ -59,7 +59,7 @@ public void run() {
List additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
additionalModules.add(new BackupModule());
- Esop.init(this, jmxSpec, hashSpec, additionalModules);
+ Esop.init(this, jmxSpec, hashSpec, request, additionalModules);
final Operation> operation = operationsService.submitOperationRequest(request);
diff --git a/core/src/main/java/com/instaclustr/esop/cli/CommitLogBackupApplication.java b/core/src/main/java/com/instaclustr/esop/cli/CommitLogBackupApplication.java
index 3757b029..b4251035 100644
--- a/core/src/main/java/com/instaclustr/esop/cli/CommitLogBackupApplication.java
+++ b/core/src/main/java/com/instaclustr/esop/cli/CommitLogBackupApplication.java
@@ -53,7 +53,7 @@ public void run() {
List additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
additionalModules.add(new CommitlogBackupModule());
- Esop.init(this, jmxSpec, hashSpec, additionalModules);
+ Esop.init(this, jmxSpec, hashSpec, request, additionalModules);
final Operation> operation = operationsService.submitOperationRequest(request);
diff --git a/core/src/main/java/com/instaclustr/esop/cli/CommitLogRestoreApplication.java b/core/src/main/java/com/instaclustr/esop/cli/CommitLogRestoreApplication.java
index 90b20a68..8b4866ba 100644
--- a/core/src/main/java/com/instaclustr/esop/cli/CommitLogRestoreApplication.java
+++ b/core/src/main/java/com/instaclustr/esop/cli/CommitLogRestoreApplication.java
@@ -49,7 +49,7 @@ public void run() {
List additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
additionalModules.add(new RestoreCommitlogModule());
- Esop.init(this, null, hashSpec, additionalModules);
+ Esop.init(this, null, hashSpec, request, additionalModules);
final Operation> operation = operationsService.submitOperationRequest(request);
diff --git a/core/src/main/java/com/instaclustr/esop/cli/Esop.java b/core/src/main/java/com/instaclustr/esop/cli/Esop.java
index c0ee9bf1..432b3bf7 100644
--- a/core/src/main/java/com/instaclustr/esop/cli/Esop.java
+++ b/core/src/main/java/com/instaclustr/esop/cli/Esop.java
@@ -12,6 +12,7 @@
import com.google.inject.Stage;
import com.instaclustr.cassandra.CassandraModule;
import com.instaclustr.esop.SPIModule;
+import com.instaclustr.esop.impl.AbstractOperationRequest;
import com.instaclustr.esop.impl.backup.BackupModules.UploadingModule;
import com.instaclustr.esop.impl.hash.HashModule;
import com.instaclustr.esop.impl.hash.HashSpec;
@@ -70,6 +71,7 @@ public static void main(String[] args, boolean exit) {
static void init(final Runnable command,
final CassandraJMXSpec jmxSpec,
final HashSpec hashSpec,
+ final AbstractOperationRequest request,
final List additionalModules) {
final List modules = new ArrayList<>();
@@ -92,7 +94,7 @@ protected void configure() {
modules.add(new ExecutorsModule());
modules.add(new UploadingModule());
modules.add(new DownloadingModule());
- modules.add(new HashModule(hashSpec));
+ modules.add(new HashModule(hashSpec, request.concurrentConnections));
modules.addAll(additionalModules);
final Injector injector = Guice.createInjector(
diff --git a/core/src/main/java/com/instaclustr/esop/cli/ListApplication.java b/core/src/main/java/com/instaclustr/esop/cli/ListApplication.java
index 79f9ea99..a7b31c36 100644
--- a/core/src/main/java/com/instaclustr/esop/cli/ListApplication.java
+++ b/core/src/main/java/com/instaclustr/esop/cli/ListApplication.java
@@ -54,7 +54,7 @@ public void run() {
List additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
additionalModules.add(new ListModule());
- Esop.init(this, jmxSpec, new HashSpec(), additionalModules);
+ Esop.init(this, jmxSpec, new HashSpec(), request, additionalModules);
final Operation> operation = operationsService.submitOperationRequest(request);
diff --git a/core/src/main/java/com/instaclustr/esop/cli/RemoveBackupApplication.java b/core/src/main/java/com/instaclustr/esop/cli/RemoveBackupApplication.java
index 9720d91b..e2abdac2 100644
--- a/core/src/main/java/com/instaclustr/esop/cli/RemoveBackupApplication.java
+++ b/core/src/main/java/com/instaclustr/esop/cli/RemoveBackupApplication.java
@@ -78,7 +78,7 @@ public void run() {
List additionalModules = new ArrayList<>(Esop.getStorageSpecificModules());
additionalModules.add(new RemoveBackupModule());
- Esop.init(this, jmxSpec, new HashSpec(), additionalModules);
+ Esop.init(this, jmxSpec, new HashSpec(), request, additionalModules);
if (rate.value == 0) {
final Operation> operation = operationsService.submitOperationRequest(request);
diff --git a/core/src/main/java/com/instaclustr/esop/cli/RestoreApplication.java b/core/src/main/java/com/instaclustr/esop/cli/RestoreApplication.java
index 74f3aa71..ef28e71a 100644
--- a/core/src/main/java/com/instaclustr/esop/cli/RestoreApplication.java
+++ b/core/src/main/java/com/instaclustr/esop/cli/RestoreApplication.java
@@ -61,7 +61,7 @@ public void run() {
additionalModules.add(new RestoreModule());
additionalModules.add(new RestorationStrategyModule());
- Esop.init(this, jmxSpec, hashSpec, additionalModules);
+ Esop.init(this, jmxSpec, hashSpec, request, additionalModules);
final Operation> operation = operationsService.submitOperationRequest(request);
diff --git a/core/src/main/java/com/instaclustr/esop/impl/AbstractOperationRequest.java b/core/src/main/java/com/instaclustr/esop/impl/AbstractOperationRequest.java
index 4dae615b..36263e9f 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/AbstractOperationRequest.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/AbstractOperationRequest.java
@@ -22,7 +22,7 @@ public abstract class AbstractOperationRequest extends OperationRequest {
converter = StorageLocationTypeConverter.class,
description = "Location to which files will be backed up or restored from, in form " +
"cloudProvider://bucketName/clusterId/datacenterId/nodeId or file:///some/path/bucketName/clusterId/datacenterId/nodeId. " +
- "'cloudProvider' is one of 's3', 'oracle', 'ceph', 'minio', 'azure' or 'gcp'.",
+ "'cloudProvider' is one of 's3', 'azure' or 'gcp'.",
required = true)
@JsonSerialize(using = StorageLocationSerializer.class)
@JsonDeserialize(using = StorageLocationDeserializer.class)
@@ -47,9 +47,8 @@ public abstract class AbstractOperationRequest extends OperationRequest {
@JsonProperty("retry")
public RetrySpec retry = new RetrySpec();
- @Option(names = {"--cc", "--concurrent-connections"},
- description = "Number of files (or file parts) to download concurrently. Higher values will increase throughput. Default is 10.",
- defaultValue = "10"
+ @Option(names = {"--cc", "--concurrent-connections", "--parallelism"},
+ description = "Number of files (or file parts) to download / upload / hash concurrently. Higher values will increase throughput. Default is 50% of available CPUs."
)
@JsonProperty("concurrentConnections")
public Integer concurrentConnections;
@@ -60,6 +59,8 @@ public abstract class AbstractOperationRequest extends OperationRequest {
public AbstractOperationRequest() {
// for picocli
+ if (concurrentConnections == null)
+ concurrentConnections = getDefaultConcurrentConnections();
}
public AbstractOperationRequest(final StorageLocation storageLocation,
@@ -74,7 +75,7 @@ public AbstractOperationRequest(final StorageLocation storageLocation,
this.skipBucketVerification = skipBucketVerification;
this.proxySettings = proxySettings;
this.retry = retry == null ? new RetrySpec() : retry;
- this.concurrentConnections = concurrentConnections == null ? 10 : concurrentConnections;
+ this.concurrentConnections = concurrentConnections == null ? getDefaultConcurrentConnections() : concurrentConnections;
this.kmsKeyId = kmsKeyId;
}
@@ -96,5 +97,21 @@ public void validate(final Set storageProviders) {
if (storageProviders != null && !storageProviders.contains(storageLocation.storageProvider)) {
throw new IllegalStateException(format("Available storage providers: %s", Arrays.toString(storageProviders.toArray())));
}
+
+ if (concurrentConnections <= 0) {
+ throw new IllegalStateException("--parallelism must be greater than 0");
+ }
+
+ if (concurrentConnections > Runtime.getRuntime().availableProcessors()) {
+ throw new IllegalStateException("--parallelism value cannot be greater than number of available processors: "
+ + Runtime.getRuntime().availableProcessors());
+ }
+ }
+
+ /**
+ * Get default number of concurrent connections based on 50% of available processors.
+ */
+ private static int getDefaultConcurrentConnections() {
+ return Runtime.getRuntime().availableProcessors() / 2;
}
}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/SSTableUtils.java b/core/src/main/java/com/instaclustr/esop/impl/SSTableUtils.java
index 18276dfb..4abd4240 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/SSTableUtils.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/SSTableUtils.java
@@ -41,7 +41,6 @@ public class SSTableUtils {
private static final int SSTABLE_PREFIX_IDX = 1;
private static final int SSTABLE_GENERATION_IDX = 2;
private static final Pattern CHECKSUM_RE = Pattern.compile("^([a-zA-Z0-9]+).*");
- private static final HashService hashService = new HashServiceImpl(new HashSpec());
public static String sstableHash(Path path) throws IOException {
final Matcher matcher = SSTABLE_RE.matcher(path.getFileName().toString());
@@ -102,14 +101,11 @@ public static String calculateChecksum(final Path filePath) throws IOException {
public static Map> getSSTables(String keyspace,
String table,
Path snapshotDirectory,
- Path tableBackupPath,
- HashSpec hashSpec) throws IOException {
+ Path tableBackupPath) throws IOException {
if (!Files.exists(snapshotDirectory)) {
return Collections.emptyMap();
}
- final HashService hashService = new HashServiceImpl(hashSpec);
-
return Files.list(snapshotDirectory)
.flatMap(path -> {
if (isCassandra22SecIndex(path)) {
@@ -148,12 +144,11 @@ public static Map> getSSTables(String keyspace,
}
backupPath = backupPath.resolve(hash).resolve(manifestComponentFileName.getFileName());
- final String hashOfFile = hashService.hash(sstableComponent);
entries.add(new ManifestEntry(backupPath,
sstableComponent,
ManifestEntry.Type.FILE,
- hashOfFile,
+ null,
new KeyspaceTable(keyspace, table),
null));
}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/Snapshots.java b/core/src/main/java/com/instaclustr/esop/impl/Snapshots.java
index 854bb2f1..8078e8d5 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/Snapshots.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/Snapshots.java
@@ -46,8 +46,6 @@
public class Snapshots implements Cloneable {
- public static HashSpec hashSpec;
-
private final Map snapshots = new HashMap<>();
public static Snapshots of(Map snapshots) {
@@ -569,7 +567,7 @@ public static Table parse(final String keyspace, final String table, final List<
final Path tablePath = Paths.get("data").resolve(Paths.get(keyspace, table));
for (final Path path : value) {
- tb.sstables.putAll(SSTableUtils.getSSTables(keyspace, table, path, tablePath, Snapshots.hashSpec));
+ tb.sstables.putAll(SSTableUtils.getSSTables(keyspace, table, path, tablePath));
}
final Optional schemaPath = value.stream().map(p -> p.resolve("schema.cql")).filter(Files::exists).findFirst();
@@ -750,10 +748,6 @@ public static synchronized Snapshots merge(final List scannedSnapshot
}
public static synchronized Snapshots parse(final Path cassandraDir, final String snapshot) throws Exception {
- if (Snapshots.hashSpec == null) {
- Snapshots.hashSpec = new HashSpec();
- }
-
final Snapshots snapshots = new Snapshots();
final SnapshotLister lister = new SnapshotLister();
Files.walkFileTree(cassandraDir, lister);
diff --git a/core/src/main/java/com/instaclustr/esop/impl/backup/coordination/BaseBackupOperationCoordinator.java b/core/src/main/java/com/instaclustr/esop/impl/backup/coordination/BaseBackupOperationCoordinator.java
index 4de04594..5363e849 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/backup/coordination/BaseBackupOperationCoordinator.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/backup/coordination/BaseBackupOperationCoordinator.java
@@ -28,6 +28,8 @@
import com.instaclustr.esop.impl.backup.coordination.ClearSnapshotOperation.ClearSnapshotOperationRequest;
import com.instaclustr.esop.impl.backup.coordination.TakeSnapshotOperation.TakeSnapshotOperationRequest;
import com.instaclustr.esop.impl.hash.HashSpec;
+import com.instaclustr.esop.impl.hash.ParallelHashService;
+import com.instaclustr.esop.impl.hash.ParallelHashServiceImpl;
import com.instaclustr.esop.impl.interaction.CassandraSchemaVersion;
import com.instaclustr.esop.impl.interaction.CassandraTokens;
import com.instaclustr.esop.topology.CassandraClusterTopology;
@@ -126,8 +128,8 @@ public void coordinate(final Operation operation) {
new TakeSnapshotOperationRequest(request.entities, request.snapshotTag),
cassandraVersionProvider).run0();
- Snapshots.hashSpec = hashSpec;
final Snapshots snapshots = Snapshots.parse(request.dataDirs, request.snapshotTag);
+
final Optional snapshot = snapshots.get(request.snapshotTag);
if (!snapshot.isPresent()) {
@@ -139,6 +141,11 @@ public void coordinate(final Operation operation) {
manifest.setSchemaVersion(request.schemaVersion);
manifest.setTokens(tokens);
+ // Compute hashes and populate it to manifest entries
+ try (ParallelHashService parallelHashService = new ParallelHashServiceImpl(hashSpec, request.concurrentConnections)) {
+ parallelHashService.hashAndPopulate(manifest.getManifestEntries(false));
+ }
+
// manifest
final Path localManifestPath = getLocalManifestPath(request.snapshotTag);
manifest.setManifest(getManifestAsManifestEntry(localManifestPath, request));
diff --git a/core/src/main/java/com/instaclustr/esop/impl/hash/HashModule.java b/core/src/main/java/com/instaclustr/esop/impl/hash/HashModule.java
index 6ff68bda..1cbf7995 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/hash/HashModule.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/hash/HashModule.java
@@ -5,14 +5,17 @@
public class HashModule extends AbstractModule {
private final HashSpec hashSpec;
+ private final int parallelHashingThreads;
- public HashModule(final HashSpec hashSpec) {
+ public HashModule(final HashSpec hashSpec, final int parallelHashingThreads) {
this.hashSpec = hashSpec;
+ this.parallelHashingThreads = parallelHashingThreads;
}
@Override
protected void configure() {
bind(HashSpec.class).toInstance(this.hashSpec);
bind(HashService.class).toInstance(new HashServiceImpl(this.hashSpec));
+ bind(ParallelHashService.class).toProvider(() -> new ParallelHashServiceImpl(this.hashSpec, this.parallelHashingThreads));
}
}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/hash/HashService.java b/core/src/main/java/com/instaclustr/esop/impl/hash/HashService.java
index 3bf18c90..38ccaad9 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/hash/HashService.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/hash/HashService.java
@@ -14,7 +14,7 @@ public interface HashService {
void verify(Path file, String hash) throws HashVerificationException;
- class HashingException extends Exception {
+ class HashingException extends RuntimeException {
public HashingException(final String message) {
super(message);
@@ -25,7 +25,7 @@ public HashingException(final String message, final Throwable cause) {
}
}
- class HashVerificationException extends Exception {
+ class HashVerificationException extends RuntimeException {
public HashVerificationException(final String message) {
super(message);
diff --git a/core/src/main/java/com/instaclustr/esop/impl/hash/HashServiceImpl.java b/core/src/main/java/com/instaclustr/esop/impl/hash/HashServiceImpl.java
index e777e3df..a122c967 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/hash/HashServiceImpl.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/hash/HashServiceImpl.java
@@ -1,9 +1,9 @@
package com.instaclustr.esop.impl.hash;
import java.io.File;
-import java.io.FileInputStream;
-import java.io.InputStream;
+import java.nio.channels.FileChannel;
import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
import com.google.inject.Inject;
import com.instaclustr.esop.impl.ManifestEntry;
@@ -87,9 +87,9 @@ private String getHash(final File file) throws Exception
{
if (hashSpec.algorithm == HashSpec.HashAlgorithm.NONE)
return null;
- try (final InputStream is = new FileInputStream(file)) {
+ try (final FileChannel ch = FileChannel.open(file.toPath(), StandardOpenOption.READ)) {
logger.info("Getting {} hash of {} ", hashSpec.algorithm.toString(), file.getAbsolutePath());
- return hashSpec.algorithm.getHasher().getHash(is);
+ return hashSpec.algorithm.getHasher().getHash(ch);
}
}
}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/hash/HashSpec.java b/core/src/main/java/com/instaclustr/esop/impl/hash/HashSpec.java
index d4d997f6..9a0aa858 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/hash/HashSpec.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/hash/HashSpec.java
@@ -1,12 +1,16 @@
package com.instaclustr.esop.impl.hash;
-import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
import java.security.MessageDigest;
import java.util.Arrays;
+import java.util.function.BiConsumer;
import java.util.function.Supplier;
import java.util.zip.CRC32;
import java.util.zip.Checksum;
+import net.jpountz.xxhash.StreamingXXHash64;
+import net.jpountz.xxhash.XXHashFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import picocli.CommandLine;
@@ -16,6 +20,9 @@
public class HashSpec {
+ // Chunk size for reading files for hashing
+ private static final int CHUNK_SIZE = 4096;
+
public HashSpec(final HashAlgorithm algorithm) {
this.algorithm = algorithm;
}
@@ -25,9 +32,9 @@ public HashSpec() {
}
@Option(names = {"--hash-algorithm"},
- description = "Algorithm to use for hashing of SSTables and files to upload / download. For skipping, use NONE.",
- defaultValue = "SHA-256",
- converter = HashAlgorithmConverter.class)
+ description = "Algorithm to use for hashing of SSTables and files to upload / download. For skipping, use NONE.",
+ defaultValue = "SHA-256",
+ converter = HashAlgorithmConverter.class)
public HashAlgorithm algorithm;
private static class HashAlgorithmConverter implements CommandLine.ITypeConverter {
@@ -40,34 +47,35 @@ public HashAlgorithm convert(final String value) {
public interface Hasher {
- String getHash(InputStream is) throws Exception;
+ default void doHashInternal(ReadableByteChannel ch, BiConsumer consumer) throws Exception
+ {
+ ByteBuffer bb = ByteBuffer.allocate(CHUNK_SIZE);
+ int bytesRead = 0;
+ while ((bytesRead = ch.read(bb)) != -1) {
+ bb.flip();
+ consumer.accept(bb, bytesRead);
+ bb.clear();
+ }
+ }
+
+ String getHash(ReadableByteChannel ch) throws Exception;
String getHash(byte[] digest) throws Exception;
}
private static class SHAHasher implements Hasher {
private final String algorithm;
+
public SHAHasher(String algorithm) {
this.algorithm = algorithm;
}
@Override
- public String getHash(InputStream is) throws Exception
+ public String getHash(ReadableByteChannel ch) throws Exception
{
final MessageDigest digest = MessageDigest.getInstance(algorithm);
-
- // Create byte array to read data in chunks
- byte[] byteArray = new byte[1024];
- int bytesCount = 0;
-
- // Read file data and update in message digest
- while ((bytesCount = is.read(byteArray)) != -1) {
- digest.update(byteArray, 0, bytesCount);
- }
-
- byte[] bytes = digest.digest();
-
- return getHash(bytes);
+ doHashInternal(ch, (buffer, ignore) -> digest.update(buffer));
+ return getHash(digest.digest());
}
@Override
@@ -84,7 +92,7 @@ public String getHash(byte[] digest) throws Exception {
public static class NoOp implements Hasher {
@Override
- public String getHash(InputStream is) throws Exception {
+ public String getHash(ReadableByteChannel ch) throws Exception {
return null;
}
@@ -96,17 +104,10 @@ public String getHash(byte[] digest) throws Exception {
public static class CRCHasher implements Hasher {
@Override
- public String getHash(InputStream is) throws Exception
+ public String getHash(ReadableByteChannel ch) throws Exception
{
- byte[] byteArray = new byte[1024];
- int bytesCount = 0;
-
Checksum checksum = new CRC32();
-
- while ((bytesCount = is.read(byteArray)) != -1) {
- checksum.update(byteArray, 0, bytesCount);
- }
-
+ doHashInternal(ch, (buffer, ignored) -> checksum.update(buffer));
return Long.toString(checksum.getValue());
}
@@ -116,8 +117,29 @@ public String getHash(byte[] digest) throws Exception {
}
}
+ /**
+ * Wraps the xxHash64 algorithm. Used for fast hashing of large files as an alternative to SHA-256.
+ */
+ public static class XXHasher implements Hasher {
+
+ @Override
+ public String getHash(ReadableByteChannel ch) throws Exception {
+ try (StreamingXXHash64 xxHash64 = XXHashFactory.fastestJavaInstance().newStreamingHash64(0)) {
+ doHashInternal(ch, (buffer, bytesRead) -> xxHash64.update(buffer.array(), 0, bytesRead));
+ return Long.toHexString(xxHash64.getValue());
+ }
+ }
+
+ @Override
+ public String getHash(final byte[] digest) throws Exception {
+ // TODO do we actually need this?
+ throw new UnsupportedOperationException();
+ }
+ }
+
public enum HashAlgorithm {
SHA_256("SHA-256", () -> new SHAHasher("SHA-256")),
+ XXHASH64("xxHash64", () -> new XXHasher()),
CRC("CRC", () -> new CRCHasher()),
NONE("NONE", () -> new NoOp());
@@ -146,17 +168,17 @@ public static HashAlgorithm parse(final String value) {
}
for (final HashAlgorithm algorithm : HashAlgorithm.values()) {
- if (algorithm.name.equals(value)) {
+ if (algorithm.name.equalsIgnoreCase(value)) {
return algorithm;
}
}
logger.info(format("Unable to parse hash algorithm for value '%s', possible algorithms: %s, returning default algorithm %s",
- value,
- Arrays.toString(HashAlgorithm.values()),
- HashAlgorithm.DEFAULT_ALGORITHM));
+ value,
+ Arrays.toString(HashAlgorithm.values()),
+ HashAlgorithm.DEFAULT_ALGORITHM));
return HashAlgorithm.DEFAULT_ALGORITHM;
}
}
-}
+}
\ No newline at end of file
diff --git a/core/src/main/java/com/instaclustr/esop/impl/hash/ParallelHashService.java b/core/src/main/java/com/instaclustr/esop/impl/hash/ParallelHashService.java
new file mode 100644
index 00000000..c677df48
--- /dev/null
+++ b/core/src/main/java/com/instaclustr/esop/impl/hash/ParallelHashService.java
@@ -0,0 +1,23 @@
+package com.instaclustr.esop.impl.hash;
+
+import java.util.List;
+
+import com.instaclustr.esop.impl.ManifestEntry;
+
+
+/**
+ * Service for hashing sstable components in parallel.
+ */
+public interface ParallelHashService extends HashService, AutoCloseable {
+
+ /**
+ * Hashes and populates the hash field of the provided manifest entries in parallel. Blocks until operation is completed or failed.
+ */
+ void hashAndPopulate(List manifestEntries);
+
+ /**
+ * Verifies all manifest entries in parallel, if verification of any ManifestEntry fails it stops left submitted verifications.
+ * Blocks until operation is completed or failed.
+ */
+ void verifyAll(final List manifestEntries) throws HashVerificationException;
+}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/hash/ParallelHashServiceImpl.java b/core/src/main/java/com/instaclustr/esop/impl/hash/ParallelHashServiceImpl.java
new file mode 100644
index 00000000..b6d0e62a
--- /dev/null
+++ b/core/src/main/java/com/instaclustr/esop/impl/hash/ParallelHashServiceImpl.java
@@ -0,0 +1,79 @@
+package com.instaclustr.esop.impl.hash;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ForkJoinPool;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.instaclustr.esop.impl.ManifestEntry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Implementation of ParallelHashService that uses ForkJoinPool for parallel hashing and verification.
+ * Implements AutoCloseable to ensure proper shutdown of the ForkJoinPool.
+ */
+public class ParallelHashServiceImpl extends HashServiceImpl implements ParallelHashService, AutoCloseable {
+ private static final Logger logger = LoggerFactory.getLogger(ParallelHashServiceImpl.class);
+ private final ForkJoinPool forkJoinPool;
+
+ public ParallelHashServiceImpl(final HashSpec hashSpec, final int parallelHashingThreads) {
+ super(hashSpec);
+ forkJoinPool = new ForkJoinPool(parallelHashingThreads);
+ }
+
+ @VisibleForTesting
+ // For testing purposes only
+ public ParallelHashServiceImpl(final HashSpec hashSpec, ForkJoinPool forkJoinPool) {
+ super(hashSpec);
+ this.forkJoinPool = forkJoinPool;
+ }
+
+ @Override
+ public void hashAndPopulate(final List manifestEntries) {
+ logger.info("Starting parallel hashing of manifest entries using {} threads.", forkJoinPool.getParallelism());
+ try {
+ forkJoinPool.submit(() -> manifestEntries.parallelStream().forEach(entry -> entry.hash = hash(entry))).get();
+ } catch (Exception e) {
+ throw new HashingException("Hashing failed for one or more manifest entries.", e);
+ }
+ }
+
+ @Override
+ public void verifyAll(final List manifestEntries) throws HashVerificationException {
+ logger.info("Starting parallel verification of manifest entries using {} threads.", forkJoinPool.getParallelism());
+
+ List> futures = new ArrayList<>();
+ for (final ManifestEntry manifestEntry : manifestEntries) {
+ CompletableFuture future = CompletableFuture.runAsync(() -> {
+ try {
+ verify(manifestEntry);
+ } catch (Exception e) {
+ logger.error("Hash verification failed, reason:", e);
+ // Cancel all other futures if one fails. Hashing operations should be interruptable in case if we want to stop them prematurely.
+ cancelAll(futures);
+ throw e;
+ }
+ }, forkJoinPool);
+ futures.add(future);
+ }
+
+ try {
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+ } catch (Exception e) {
+ throw new HashVerificationException("Hash verification failed for one or more manifest entries.", e);
+ }
+ }
+
+ private static void cancelAll(List> futures) {
+ for (CompletableFuture future : futures) {
+ future.cancel(true);
+ }
+ }
+
+ @Override
+ public void close() {
+ forkJoinPool.shutdown();
+ }
+}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/restore/RestorationPhase.java b/core/src/main/java/com/instaclustr/esop/impl/restore/RestorationPhase.java
index 818aa4a5..a1bfadc9 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/restore/RestorationPhase.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/restore/RestorationPhase.java
@@ -507,9 +507,11 @@ public void execute() throws RestorationPhaseException {
final DataVerification dataVerification = new DataVerification(ctxt).verify(manifest, databaseEntitiesToVerify);
if (dataVerification.hasErrors()) {
- throw new RestorationPhaseException("Some local files were corrupted or they are missing, please consult the logs to see the details.");
+ throw new RestorationPhaseException("Some local files were corrupted or they are missing, "
+ + "please consult the logs to see the details" + dataVerification.toString());
}
+
final List downloadedFiles = CassandraData.list(ctxt.operation.request.importing.sourceDir);
final PathSSTableClassifier pathSSTableClassifier = new PathSSTableClassifier(ctxt.operation.request);
final Map> classifiedDownloadedFiles = pathSSTableClassifier.classify(downloadedFiles);
@@ -717,18 +719,19 @@ public static final class DataVerification {
private final RestorationContext ctxt;
public final List nonExistingFiles = new ArrayList<>();
- public final List corruptedFiles = new ArrayList<>();
+ public Throwable exceptionDuringVerification;
public DataVerification(final RestorationContext ctxt) {
this.ctxt = ctxt;
}
public boolean hasErrors() {
- return !nonExistingFiles.isEmpty() || !corruptedFiles.isEmpty();
+ return !nonExistingFiles.isEmpty() || exceptionDuringVerification != null;
}
public DataVerification verify(final Manifest manifest, final DatabaseEntities entities) {
final List entries = manifest.getManifestFiles(entities, false, false, false, false);
+ final List entriesToVerify = new ArrayList<>();
for (final ManifestEntry entry : entries) {
if (!Files.exists(entry.localFile)) {
@@ -738,15 +741,16 @@ public DataVerification verify(final Manifest manifest, final DatabaseEntities e
}
if (entry.hash != null) {
- try {
- this.ctxt.hashService.verify(entry.localFile, entry.hash);
- } catch (final Exception ex) {
- logger.error(ex.getMessage());
- corruptedFiles.add(entry.localFile.toString());
- }
+ entriesToVerify.add(entry);
}
}
+ try {
+ ctxt.hashService.verifyAll(entriesToVerify);
+ } catch (Exception e) {
+ exceptionDuringVerification = e;
+ }
+
return this;
}
@@ -754,7 +758,7 @@ public DataVerification verify(final Manifest manifest, final DatabaseEntities e
public String toString() {
return MoreObjects.toStringHelper(this)
.add("nonExistingFiles", nonExistingFiles)
- .add("corruptedFiles", corruptedFiles)
+ .add("exceptionDuringVerification", exceptionDuringVerification)
.toString();
}
}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/AbstractRestorationStrategy.java b/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/AbstractRestorationStrategy.java
index 09ab2d8b..7d2c5b4f 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/AbstractRestorationStrategy.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/AbstractRestorationStrategy.java
@@ -7,7 +7,7 @@
import com.google.inject.Provider;
import com.instaclustr.cassandra.CassandraVersion;
import com.instaclustr.esop.guice.BucketServiceFactory;
-import com.instaclustr.esop.impl.hash.HashService;
+import com.instaclustr.esop.impl.hash.ParallelHashService;
import com.instaclustr.esop.impl.restore.DownloadTracker;
import com.instaclustr.esop.impl.restore.RestorationPhase;
import com.instaclustr.esop.impl.restore.RestorationPhase.ClusterHealthCheckPhase;
@@ -32,14 +32,14 @@ public abstract class AbstractRestorationStrategy implements RestorationStrategy
protected final ObjectMapper objectMapper;
protected final DownloadTracker downloadTracker;
protected final Map bucketServiceFactoryMap;
- protected final HashService hashService;
+ protected final ParallelHashService hashService;
public AbstractRestorationStrategy(final CassandraJMXService cassandraJMXService,
final Provider cassandraVersion,
final ObjectMapper objectMapper,
final DownloadTracker downloadTracker,
final Map bucketServiceFactoryMap,
- final HashService hashService) {
+ final ParallelHashService hashService) {
this.cassandraJMXService = cassandraJMXService;
this.cassandraVersion = cassandraVersion;
this.objectMapper = objectMapper;
diff --git a/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/HardlinkingRestorationStrategy.java b/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/HardlinkingRestorationStrategy.java
index 44d2db19..cd161f5e 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/HardlinkingRestorationStrategy.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/HardlinkingRestorationStrategy.java
@@ -7,7 +7,7 @@
import com.google.inject.Provider;
import com.instaclustr.cassandra.CassandraVersion;
import com.instaclustr.esop.guice.BucketServiceFactory;
-import com.instaclustr.esop.impl.hash.HashService;
+import com.instaclustr.esop.impl.hash.ParallelHashService;
import com.instaclustr.esop.impl.restore.DownloadTracker;
import com.instaclustr.esop.impl.restore.RestorationPhase;
import com.instaclustr.esop.impl.restore.RestorationPhase.CleaningPhase;
@@ -50,7 +50,7 @@ public HardlinkingRestorationStrategy(final CassandraJMXService cassandraJMXServ
final ObjectMapper objectMapper,
final DownloadTracker downloadTracker,
final Map bucketServiceFactoryMap,
- final HashService hashService) {
+ final ParallelHashService hashService) {
super(cassandraJMXService, cassandraVersion, objectMapper, downloadTracker, bucketServiceFactoryMap, hashService);
}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/ImportingRestorationStrategy.java b/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/ImportingRestorationStrategy.java
index 541cdef7..5696ee58 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/ImportingRestorationStrategy.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/ImportingRestorationStrategy.java
@@ -7,7 +7,7 @@
import com.google.inject.Provider;
import com.instaclustr.cassandra.CassandraVersion;
import com.instaclustr.esop.guice.BucketServiceFactory;
-import com.instaclustr.esop.impl.hash.HashService;
+import com.instaclustr.esop.impl.hash.ParallelHashService;
import com.instaclustr.esop.impl.restore.DownloadTracker;
import com.instaclustr.esop.impl.restore.RestorationPhase;
import com.instaclustr.esop.impl.restore.RestorationPhase.CleaningPhase;
@@ -49,7 +49,7 @@ public ImportingRestorationStrategy(final CassandraJMXService cassandraJMXServic
final ObjectMapper objectMapper,
final DownloadTracker downloadTracker,
final Map bucketServiceFactoryMap,
- final HashService hashService) {
+ final ParallelHashService hashService) {
super(cassandraJMXService, cassandraVersion, objectMapper, downloadTracker, bucketServiceFactoryMap, hashService);
}
diff --git a/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/RestorationContext.java b/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/RestorationContext.java
index 5416ce11..a2cc66bc 100644
--- a/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/RestorationContext.java
+++ b/core/src/main/java/com/instaclustr/esop/impl/restore/strategy/RestorationContext.java
@@ -6,7 +6,7 @@
import com.instaclustr.cassandra.CassandraVersion;
import com.instaclustr.esop.guice.BucketServiceFactory;
import com.instaclustr.esop.impl.CassandraData;
-import com.instaclustr.esop.impl.hash.HashService;
+import com.instaclustr.esop.impl.hash.ParallelHashService;
import com.instaclustr.esop.impl.restore.DownloadTracker;
import com.instaclustr.esop.impl.restore.RestorationPhase.RestorationPhaseType;
import com.instaclustr.esop.impl.restore.RestoreOperationRequest;
@@ -26,5 +26,5 @@ public class RestorationContext {
public DownloadTracker downloadTracker;
public Map bucketServiceFactoryMap;
public CassandraData cassandraData;
- public HashService hashService;
+ public ParallelHashService hashService;
}
diff --git a/core/src/test/java/com/instaclustr/esop/backup/BackupRestoreTest.java b/core/src/test/java/com/instaclustr/esop/backup/BackupRestoreTest.java
index 7a36ec5c..e1f1f21f 100644
--- a/core/src/test/java/com/instaclustr/esop/backup/BackupRestoreTest.java
+++ b/core/src/test/java/com/instaclustr/esop/backup/BackupRestoreTest.java
@@ -85,11 +85,11 @@ public void testSSTableLister() throws Exception {
final String keyspace = "keyspace1";
final String table1 = "table1";
final Path table1Path = tempDirs.get(testFileConfig.cassandraVersion.toString()).resolve("data/" + keyspace + "/" + table1);
- Map> sstables = SSTableUtils.getSSTables(keyspace, table1, table1Path, backupRoot.resolve(table1Path.getFileName()), new HashSpec());
+ Map> sstables = SSTableUtils.getSSTables(keyspace, table1, table1Path, backupRoot.resolve(table1Path.getFileName()));
final String table2 = "table2";
final Path table2Path = tempDirs.get(testFileConfig.cassandraVersion.toString()).resolve("data/" + keyspace + "/" + table2);
- sstables.putAll(SSTableUtils.getSSTables(keyspace, table2, table2Path, backupRoot.resolve(table2Path.getFileName()), new HashSpec()));
+ sstables.putAll(SSTableUtils.getSSTables(keyspace, table2, table2Path, backupRoot.resolve(table2Path.getFileName())));
Map manifestMap = new HashMap<>();
for (ManifestEntry e : sstables.values().stream().flatMap(Collection::stream).collect(Collectors.toList())) {
diff --git a/core/src/test/java/com/instaclustr/esop/backup/HashServiceTest.java b/core/src/test/java/com/instaclustr/esop/backup/HashServiceTest.java
index 1f7cd1e3..043e2081 100644
--- a/core/src/test/java/com/instaclustr/esop/backup/HashServiceTest.java
+++ b/core/src/test/java/com/instaclustr/esop/backup/HashServiceTest.java
@@ -1,21 +1,65 @@
-package com.instaclustr.esop.backup;
+ package com.instaclustr.esop.backup;
-import java.io.File;
-import java.nio.file.Files;
+ import java.net.URL;
+ import java.nio.file.Path;
-import com.instaclustr.esop.impl.hash.HashService;
-import com.instaclustr.esop.impl.hash.HashServiceImpl;
-import com.instaclustr.esop.impl.hash.HashSpec;
-import org.junit.jupiter.api.Test;
+ import com.instaclustr.esop.impl.hash.HashService;
+ import com.instaclustr.esop.impl.hash.HashServiceImpl;
+ import com.instaclustr.esop.impl.hash.HashSpec;
+ import org.junit.jupiter.api.BeforeAll;
+ import org.junit.jupiter.api.Test;
-public class HashServiceTest {
+ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+ import static org.junit.jupiter.api.Assertions.assertNotNull;
- @Test
- public void testHashing() throws Exception {
- final File f = File.createTempFile("hashingTest", ".tmp");
- Files.write(f.toPath(), "".getBytes());
- final HashService hashService = new HashServiceImpl(new HashSpec());
- hashService.verify(f.toPath(), hashService.hash(f.toPath()));
- }
+ public class HashServiceTest {
+
+ private static final String HASH_TEST_FILE_LOCATION = "hash/hash-test-file-12K";
+ private static Path testFilePath;
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ URL resourceUrl = HashServiceTest.class.getClassLoader().getResource(HASH_TEST_FILE_LOCATION);
+ assertNotNull(resourceUrl, "Test file resource not found: " + HASH_TEST_FILE_LOCATION);
+ testFilePath = Path.of(resourceUrl.toURI().getPath()).toAbsolutePath();
+ }
+
+ // Generated with: sha256sum
+ private static final String EXPECTED_SHA256_HASH = "01d8740e8d0b16d0468324a7952f483c9f360529966b29b6cd1bf81ca2988c5b";
+ // Generated with: xxhsum -H1 which is xxhash64
+ private static final String EXPECTED_XXHASH64_HASH = "ce26d6e69ac4d755";
+ // Generated with: echo $((16#$(crc32 ./core/src/test/resources/hash/hash-test-file-12K)))\
+ // to install crc32 tool: apt install libarchive-zip-perl
+ private static final String EXPECTED_CRC32_HASH = "2657330654";
+
+ @Test
+ public void testHashing_DefaultAlgorithm() {
+ testHashing(new HashSpec(), EXPECTED_SHA256_HASH);
+ }
+
+ @Test
+ public void testHashing_SHA256() {
+ testHashing(new HashSpec(HashSpec.HashAlgorithm.SHA_256), EXPECTED_SHA256_HASH);
+ }
-}
+ @Test
+ public void testHashing_xxHash64() {
+ testHashing(new HashSpec(HashSpec.HashAlgorithm.XXHASH64), EXPECTED_XXHASH64_HASH);
+ }
+
+ @Test
+ public void testHashing_Crc32() {
+ testHashing(new HashSpec(HashSpec.HashAlgorithm.CRC), EXPECTED_CRC32_HASH);
+ }
+
+
+ @Test
+ public void testHashing_None() {
+ testHashing(new HashSpec(HashSpec.HashAlgorithm.NONE), null);
+ }
+
+ private void testHashing(HashSpec hashSpec, String expectedHash) {
+ final HashService hashService = new HashServiceImpl(hashSpec);
+ assertDoesNotThrow(() -> hashService.verify(testFilePath, expectedHash));
+ }
+ }
diff --git a/core/src/test/java/com/instaclustr/esop/backup/ParallelHashServiceTest.java b/core/src/test/java/com/instaclustr/esop/backup/ParallelHashServiceTest.java
new file mode 100644
index 00000000..1433c506
--- /dev/null
+++ b/core/src/test/java/com/instaclustr/esop/backup/ParallelHashServiceTest.java
@@ -0,0 +1,134 @@
+package com.instaclustr.esop.backup;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ForkJoinPool;
+import java.util.stream.Collectors;
+
+import com.instaclustr.esop.impl.ManifestEntry;
+import com.instaclustr.esop.impl.hash.HashService;
+import com.instaclustr.esop.impl.hash.HashService.HashingException;
+import com.instaclustr.esop.impl.hash.HashSpec;
+import com.instaclustr.esop.impl.hash.ParallelHashService;
+import com.instaclustr.esop.impl.hash.ParallelHashServiceImpl;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public class ParallelHashServiceTest {
+ private static final int THREAD_COUNT = 3;
+ private static Path tempDir;
+ private static List testManifestEntries = new ArrayList<>();
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ tempDir = Files.createTempDirectory("paralell-files-hash-test");
+ for (int i = 0; i < THREAD_COUNT * 3; i++) {
+ Path testFilePath = Files.createTempFile(tempDir, "test", "test");
+ // writing file path as content to have some data
+ Files.writeString(testFilePath, testFilePath.toString(), StandardCharsets.UTF_8);
+ }
+ }
+
+ @BeforeEach
+ public void prepareManifestEntries() throws Exception {
+ testManifestEntries.clear();
+ Files.list(tempDir).forEach(path -> {
+ testManifestEntries.add(new ManifestEntry(
+ null,
+ path,
+ ManifestEntry.Type.FILE,
+ null,
+ null));
+ });
+ }
+
+ @Test
+ public void testHashAndPopulate() {
+ try (ParallelHashService parallelHashService = new ParallelHashServiceImpl(new HashSpec(HashSpec.HashAlgorithm.XXHASH64), THREAD_COUNT)) {
+ parallelHashService.hashAndPopulate(testManifestEntries);
+
+ for (ManifestEntry entry : testManifestEntries) {
+ // Hash should be populated for each ManifestEntry
+ assertDoesNotThrow(() -> parallelHashService.verify(entry));
+ }
+ } catch (Exception e) {
+ fail("Should not throw exception during initialization of ParallelHashService", e);
+ }
+ }
+
+ @Test
+ public void testHashAndPopulateThrowsExceptionDuringExecution() {
+ List faultyManifests = faultyManifestEntriesOf(testManifestEntries);
+
+ try (ParallelHashService parallelHashService = new ParallelHashServiceImpl(new HashSpec(HashSpec.HashAlgorithm.XXHASH64), THREAD_COUNT)) {
+ assertThrowsExactly(HashingException.class, () -> parallelHashService.hashAndPopulate(faultyManifests));
+ } catch (Exception e) {
+ fail("Should not throw exception during initialization of ParallelHashService", e);
+ }
+ }
+
+ @Test
+ public void testVerifyAll() {
+ try (ParallelHashService parallelHashService = new ParallelHashServiceImpl(new HashSpec(HashSpec.HashAlgorithm.XXHASH64), THREAD_COUNT)) {
+ // First, hash and populate the entries
+ parallelHashService.hashAndPopulate(testManifestEntries);
+
+ // Now verify all entries
+ parallelHashService.verifyAll(testManifestEntries);
+ } catch (Exception e) {
+ fail("Should not throw", e);
+ }
+ }
+
+ @Test
+ public void testVerifyAllExceptionDuringExecution() {
+ ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
+
+ try (ParallelHashService parallelHashService = new ParallelHashServiceImpl(new HashSpec(HashSpec.HashAlgorithm.XXHASH64), forkJoinPool)) {
+
+ // First, hash and populate the entries
+ parallelHashService.hashAndPopulate(testManifestEntries);
+ // Create faulty manifest entries
+ List copiedEntries = copyManifestEntriesOf(testManifestEntries);
+ for (ManifestEntry entry : copiedEntries) {
+ entry.hash = "invalid-hash-value";
+ }
+
+ // Now verify all entries
+ assertThrowsExactly(HashService.HashVerificationException.class, () -> {
+ parallelHashService.verifyAll(copiedEntries);
+ });
+
+ assertTrue(forkJoinPool.getRunningThreadCount() < THREAD_COUNT, "ForkJoinPool should have no queued tasks after verification failure");
+ }
+ catch (Exception e) {
+ fail("Should not throw exception during initialization of ParallelHashService", e);
+ }
+ }
+
+ private static List faultyManifestEntriesOf(List originalEntries) {
+ List faultyManifests = copyManifestEntriesOf(testManifestEntries);
+ faultyManifests.forEach(m -> m.localFile = Path.of("non-existing-path"));
+ return faultyManifests;
+ }
+
+ private static List copyManifestEntriesOf(List originalEntries) {
+ return originalEntries.stream().map(
+ e -> new ManifestEntry(
+ null,
+ e.localFile,
+ e.type,
+ e.hash,
+ null)
+ ).collect(Collectors.toList());
+ }
+}
diff --git a/core/src/test/java/com/instaclustr/esop/backup/embedded/AbstractBackupTest.java b/core/src/test/java/com/instaclustr/esop/backup/embedded/AbstractBackupTest.java
index 5c2263d4..c6142315 100644
--- a/core/src/test/java/com/instaclustr/esop/backup/embedded/AbstractBackupTest.java
+++ b/core/src/test/java/com/instaclustr/esop/backup/embedded/AbstractBackupTest.java
@@ -81,7 +81,7 @@ public String getCassandraVersion() {
add(new RestorationStrategyModule());
add(new ListModule());
add(new RemoveBackupModule());
- add(new HashModule(new HashSpec()));
+ add(new HashModule(new HashSpec(), Runtime.getRuntime().availableProcessors()/2));
try {
add(new CassandraModule());
diff --git a/core/src/test/resources/hash/hash-test-file-12K b/core/src/test/resources/hash/hash-test-file-12K
new file mode 100644
index 00000000..bd73d20c
--- /dev/null
+++ b/core/src/test/resources/hash/hash-test-file-12K
@@ -0,0 +1,262 @@
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
+TEST DATA FOR HASHING TESTS PURPOSES ONLY.
\ No newline at end of file
diff --git a/gcp/pom.xml b/gcp/pom.xml
index b95c0200..08daf853 100644
--- a/gcp/pom.xml
+++ b/gcp/pom.xml
@@ -6,12 +6,12 @@
com.instaclustr
esop-parent
- 4.0.1
+ 4.0.2-SNAPSHOT
../pom.xml
esop-gcp
- 4.0.1
+ 4.0.2-SNAPSHOT
esop-gcp
Backup and restoration tooling for Cassandra for GCP
diff --git a/pom.xml b/pom.xml
index 76ab5aca..9b08fa8c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -5,7 +5,7 @@
com.instaclustr
esop-parent
- 4.0.1
+ 4.0.2-SNAPSHOT
pom
esop-parent
diff --git a/s3/pom.xml b/s3/pom.xml
index 1fdd9812..99b86913 100644
--- a/s3/pom.xml
+++ b/s3/pom.xml
@@ -6,12 +6,12 @@
com.instaclustr
esop-parent
- 4.0.1
+ 4.0.2-SNAPSHOT
../pom.xml
esop-s3
- 4.0.1
+ 4.0.2-SNAPSHOT
esop-s3
Backup and restoration tooling for Cassandra for AWS S3