| /* |
| * 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.spark.data; |
| |
| import java.io.IOException; |
| import java.io.ObjectInputStream; |
| import java.io.ObjectOutputStream; |
| import java.io.Serializable; |
| import java.math.BigInteger; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.CompletionStage; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.TimeUnit; |
| import java.util.function.Function; |
| import java.util.stream.Collectors; |
| import java.util.stream.Stream; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import com.google.common.cache.Cache; |
| import com.google.common.cache.CacheBuilder; |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.collect.Range; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.esotericsoftware.kryo.Kryo; |
| import com.esotericsoftware.kryo.io.Input; |
| import com.esotericsoftware.kryo.io.Output; |
| import org.apache.cassandra.bridge.BigNumberConfig; |
| import org.apache.cassandra.bridge.BigNumberConfigImpl; |
| import org.apache.cassandra.bridge.CassandraBridge; |
| import org.apache.cassandra.bridge.CassandraBridgeFactory; |
| import org.apache.cassandra.bridge.CassandraVersion; |
| import org.apache.cassandra.clients.ExecutorHolder; |
| import org.apache.cassandra.clients.Sidecar; |
| import org.apache.cassandra.clients.SidecarInstanceImpl; |
| import org.apache.cassandra.secrets.SslConfig; |
| import org.apache.cassandra.secrets.SslConfigSecretsProvider; |
| import org.apache.cassandra.sidecar.client.SidecarClient; |
| import org.apache.cassandra.sidecar.client.SidecarInstance; |
| import org.apache.cassandra.sidecar.client.SimpleSidecarInstancesProvider; |
| import org.apache.cassandra.sidecar.client.exception.RetriesExhaustedException; |
| import org.apache.cassandra.sidecar.common.NodeSettings; |
| import org.apache.cassandra.sidecar.common.data.ListSnapshotFilesResponse; |
| import org.apache.cassandra.sidecar.common.data.RingResponse; |
| import org.apache.cassandra.sidecar.common.data.SchemaResponse; |
| import org.apache.cassandra.spark.config.SchemaFeature; |
| import org.apache.cassandra.spark.config.SchemaFeatureSet; |
| import org.apache.cassandra.spark.data.partitioner.CassandraInstance; |
| import org.apache.cassandra.spark.data.partitioner.CassandraRing; |
| import org.apache.cassandra.spark.data.partitioner.ConsistencyLevel; |
| import org.apache.cassandra.spark.data.partitioner.Partitioner; |
| import org.apache.cassandra.spark.data.partitioner.TokenPartitioner; |
| import org.apache.cassandra.spark.sparksql.LastModifiedTimestampDecorator; |
| import org.apache.cassandra.spark.sparksql.RowBuilder; |
| import org.apache.cassandra.spark.stats.Stats; |
| import org.apache.cassandra.spark.utils.CqlUtils; |
| import org.apache.cassandra.spark.utils.ReaderTimeProvider; |
| import org.apache.cassandra.spark.utils.ScalaFunctions; |
| import org.apache.cassandra.spark.utils.ThrowableUtils; |
| import org.apache.cassandra.spark.utils.TimeProvider; |
| import org.apache.cassandra.spark.validation.CassandraValidation; |
| import org.apache.cassandra.spark.validation.SidecarValidation; |
| import org.apache.cassandra.spark.validation.StartupValidatable; |
| import org.apache.cassandra.spark.validation.StartupValidator; |
| import org.apache.spark.sql.types.DataType; |
| import org.apache.spark.util.ShutdownHookManager; |
| import org.jetbrains.annotations.NotNull; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.cassandra.spark.utils.Properties.NODE_STATUS_NOT_CONSIDERED; |
| |
| public class CassandraDataLayer extends PartitionedDataLayer implements StartupValidatable, Serializable |
| { |
| private static final long serialVersionUID = -9038926850642710787L; |
| |
| public static final Logger LOGGER = LoggerFactory.getLogger(CassandraDataLayer.class); |
| private static final Cache<String, CompletableFuture<List<SSTable>>> SNAPSHOT_CACHE = |
| CacheBuilder.newBuilder() |
| .expireAfterAccess(15, TimeUnit.MINUTES) |
| .maximumSize(128) |
| .build(); |
| |
| protected String snapshotName; |
| protected boolean quoteIdentifiers; |
| protected String keyspace; |
| protected String table; |
| protected String maybeQuotedKeyspace; |
| protected String maybeQuotedTable; |
| protected CassandraBridge bridge; |
| protected Set<? extends SidecarInstance> clusterConfig; |
| protected TokenPartitioner tokenPartitioner; |
| protected Map<String, AvailabilityHint> availabilityHints; |
| protected Sidecar.ClientConfig sidecarClientConfig; |
| protected Map<String, BigNumberConfigImpl> bigNumberConfigMap; |
| protected boolean enableStats; |
| protected boolean readIndexOffset; |
| protected boolean useIncrementalRepair; |
| protected List<SchemaFeature> requestedFeatures; |
| protected Map<String, ReplicationFactor> rfMap; |
| @Nullable |
| protected String lastModifiedTimestampField; |
| // volatile in order to publish the reference for visibility |
| protected volatile CqlTable cqlTable; |
| protected transient TimeProvider timeProvider; |
| protected transient SidecarClient sidecar; |
| |
| private SslConfig sslConfig; |
| |
| @VisibleForTesting |
| transient Map<String, SidecarInstance> instanceMap; |
| |
| public CassandraDataLayer(@NotNull ClientConfig options, |
| @NotNull Sidecar.ClientConfig sidecarClientConfig, |
| @Nullable SslConfig sslConfig) |
| { |
| super(options.consistencyLevel(), options.datacenter()); |
| this.snapshotName = options.snapshotName(); |
| this.keyspace = options.keyspace(); |
| this.table = options.table(); |
| this.quoteIdentifiers = options.quoteIdentifiers(); |
| this.sidecarClientConfig = sidecarClientConfig; |
| this.sslConfig = sslConfig; |
| this.bigNumberConfigMap = options.bigNumberConfigMap(); |
| this.enableStats = options.enableStats(); |
| this.readIndexOffset = options.readIndexOffset(); |
| this.useIncrementalRepair = options.useIncrementalRepair(); |
| this.lastModifiedTimestampField = options.lastModifiedTimestampField(); |
| this.requestedFeatures = options.requestedFeatures(); |
| } |
| |
| // For serialization |
| @VisibleForTesting |
| // CHECKSTYLE IGNORE: Constructor with many parameters |
| protected CassandraDataLayer(@Nullable String keyspace, |
| @Nullable String table, |
| boolean quoteIdentifiers, |
| @NotNull String snapshotName, |
| @Nullable String datacenter, |
| @NotNull Sidecar.ClientConfig sidecarClientConfig, |
| @Nullable SslConfig sslConfig, |
| @NotNull CqlTable cqlTable, |
| @NotNull TokenPartitioner tokenPartitioner, |
| @NotNull CassandraVersion version, |
| @NotNull ConsistencyLevel consistencyLevel, |
| @NotNull Set<SidecarInstanceImpl> clusterConfig, |
| @NotNull Map<String, PartitionedDataLayer.AvailabilityHint> availabilityHints, |
| @NotNull Map<String, BigNumberConfigImpl> bigNumberConfigMap, |
| boolean enableStats, |
| boolean readIndexOffset, |
| boolean useIncrementalRepair, |
| @Nullable String lastModifiedTimestampField, |
| List<SchemaFeature> requestedFeatures, |
| @NotNull Map<String, ReplicationFactor> rfMap, |
| TimeProvider timeProvider) |
| { |
| super(consistencyLevel, datacenter); |
| this.snapshotName = snapshotName; |
| this.bridge = CassandraBridgeFactory.get(version); |
| this.keyspace = keyspace; |
| this.table = table; |
| this.quoteIdentifiers = quoteIdentifiers; |
| this.cqlTable = cqlTable; |
| this.tokenPartitioner = tokenPartitioner; |
| this.clusterConfig = clusterConfig; |
| this.availabilityHints = availabilityHints; |
| this.sidecarClientConfig = sidecarClientConfig; |
| this.sslConfig = sslConfig; |
| this.bigNumberConfigMap = bigNumberConfigMap; |
| this.enableStats = enableStats; |
| this.readIndexOffset = readIndexOffset; |
| this.useIncrementalRepair = useIncrementalRepair; |
| this.lastModifiedTimestampField = lastModifiedTimestampField; |
| this.requestedFeatures = requestedFeatures; |
| if (lastModifiedTimestampField != null) |
| { |
| aliasLastModifiedTimestamp(this.requestedFeatures, this.lastModifiedTimestampField); |
| } |
| this.rfMap = rfMap; |
| this.timeProvider = timeProvider; |
| this.maybeQuoteKeyspaceAndTable(); |
| this.initInstanceMap(); |
| this.startupValidate(); |
| } |
| |
| public void initialize(@NotNull ClientConfig options) |
| { |
| dialHome(options); |
| |
| timeProvider = new ReaderTimeProvider(); |
| LOGGER.info("Starting Cassandra Spark job snapshotName={} keyspace={} table={} dc={} referenceEpoch={}", |
| snapshotName, keyspace, table, datacenter, timeProvider.referenceEpochInSeconds()); |
| |
| // Load cluster config from options |
| clusterConfig = initializeClusterConfig(options); |
| initInstanceMap(); |
| |
| // Get cluster info from Cassandra Sidecar |
| int effectiveNumberOfCores; |
| try |
| { |
| effectiveNumberOfCores = initBulkReader(options); |
| } |
| catch (InterruptedException exception) |
| { |
| Thread.currentThread().interrupt(); |
| throw new RuntimeException(exception); |
| } |
| catch (ExecutionException exception) |
| { |
| throw new RuntimeException(ThrowableUtils.rootCause(exception)); |
| } |
| LOGGER.info("Initialized Cassandra Bulk Reader with effectiveNumberOfCores={}", effectiveNumberOfCores); |
| } |
| |
| private int initBulkReader(@NotNull ClientConfig options) throws ExecutionException, InterruptedException |
| { |
| Preconditions.checkArgument(keyspace != null, "Keyspace must be non-null for Cassandra Bulk Reader"); |
| Preconditions.checkArgument(table != null, "Table must be non-null for Cassandra Bulk Reader"); |
| ShutdownHookManager.addShutdownHook(org.apache.spark.util.ShutdownHookManager.TEMP_DIR_SHUTDOWN_PRIORITY(), |
| ScalaFunctions.wrapLambda(() -> shutdownHook(options))); |
| |
| NodeSettings nodeSettings = sidecar.nodeSettings().get(); |
| String cassandraVersion = getEffectiveCassandraVersionForRead(clusterConfig, nodeSettings); |
| Partitioner partitioner = Partitioner.from(nodeSettings.partitioner()); |
| bridge = CassandraBridgeFactory.get(cassandraVersion); |
| // optionally quote identifiers if the option has been set, we need an instance for the bridge |
| maybeQuoteKeyspaceAndTable(); |
| |
| // Now we can use the quoted keyspace and table if needed |
| CompletableFuture<RingResponse> ringFuture = sidecar.ring(maybeQuotedKeyspace); |
| CompletableFuture<SchemaResponse> schemaFuture = sidecar.schema(maybeQuotedKeyspace); |
| CompletableFuture<Map<String, PartitionedDataLayer.AvailabilityHint>> snapshotFuture; |
| if (options.createSnapshot()) |
| { |
| // Use create snapshot request to capture instance availability hint |
| LOGGER.info("Creating snapshot snapshotName={} keyspace={} table={} dc={}", |
| snapshotName, maybeQuotedKeyspace, maybeQuotedTable, datacenter); |
| snapshotFuture = ringFuture.thenCompose(ringResponse -> createSnapshot(options, ringResponse)); |
| } |
| else |
| { |
| snapshotFuture = CompletableFuture.completedFuture(new HashMap<>()); |
| } |
| availabilityHints = snapshotFuture.get(); |
| |
| String fullSchema = schemaFuture.get().schema(); |
| String createStmt = CqlUtils.extractTableSchema(fullSchema, keyspace, table); |
| int indexCount = CqlUtils.extractIndexCount(fullSchema, keyspace, table); |
| Set<String> udts = CqlUtils.extractUdts(fullSchema, keyspace); |
| ReplicationFactor replicationFactor = CqlUtils.extractReplicationFactor(fullSchema, keyspace); |
| rfMap = ImmutableMap.of(keyspace, replicationFactor); |
| CompletableFuture<Integer> sizingFuture = CompletableFuture.supplyAsync( |
| () -> getSizing(clusterConfig, replicationFactor, options).getEffectiveNumberOfCores(), |
| ExecutorHolder.EXECUTOR_SERVICE); |
| validateReplicationFactor(replicationFactor); |
| udts.forEach(udt -> LOGGER.info("Adding schema UDT: '{}'", udt)); |
| |
| cqlTable = bridge().buildSchema(createStmt, keyspace, replicationFactor, partitioner, udts, null, indexCount); |
| CassandraRing ring = createCassandraRingFromRing(partitioner, replicationFactor, ringFuture.get()); |
| |
| int effectiveNumberOfCores = sizingFuture.get(); |
| tokenPartitioner = new TokenPartitioner(ring, options.defaultParallelism(), effectiveNumberOfCores); |
| return effectiveNumberOfCores; |
| } |
| |
| protected void shutdownHook(ClientConfig options) |
| { |
| ClientConfig.ClearSnapshotStrategy clearSnapshotStrategy = options.clearSnapshotStrategy(); |
| if (clearSnapshotStrategy.shouldClearOnCompletion()) |
| { |
| if (options.createSnapshot()) |
| { |
| clearSnapshot(clusterConfig, options); |
| } |
| else |
| { |
| LOGGER.warn("Skipping clearing snapshot because it was not created by this job. " |
| + "Only the job that created the snapshot can clear it. " |
| + "snapshotName={} keyspace={} table={} dc={}", |
| snapshotName, keyspace, table, datacenter); |
| } |
| } |
| else if (clearSnapshotStrategy.hasTTL()) |
| { |
| LOGGER.warn("Skipping clearing snapshot because clearSnapshotStrategy '{}' is used", clearSnapshotStrategy); |
| } |
| |
| try |
| { |
| sidecar.close(); |
| } |
| catch (Exception exception) |
| { |
| LOGGER.warn("Unable to close Sidecar", exception); |
| } |
| } |
| |
| private CompletionStage<Map<String, AvailabilityHint>> createSnapshot(ClientConfig options, RingResponse ring) |
| { |
| Map<String, PartitionedDataLayer.AvailabilityHint> availabilityHints = new ConcurrentHashMap<>(ring.size()); |
| |
| // Fire off create snapshot request across the entire cluster |
| List<CompletableFuture<Void>> futures = |
| ring.stream() |
| .filter(ringEntry -> datacenter == null || datacenter.equals(ringEntry.datacenter())) |
| .map(ringEntry -> { |
| PartitionedDataLayer.AvailabilityHint hint = |
| PartitionedDataLayer.AvailabilityHint.fromState(ringEntry.status(), ringEntry.state()); |
| |
| CompletableFuture<PartitionedDataLayer.AvailabilityHint> createSnapshotFuture; |
| if (NODE_STATUS_NOT_CONSIDERED.contains(ringEntry.state())) |
| { |
| LOGGER.warn("Skip snapshot creating when node is joining or down " |
| + "snapshotName={} keyspace={} table={} datacenter={} fqdn={} status={} state={}", |
| snapshotName, maybeQuotedKeyspace, maybeQuotedTable, datacenter, ringEntry.fqdn(), ringEntry.status(), ringEntry.state()); |
| createSnapshotFuture = CompletableFuture.completedFuture(hint); |
| } |
| else |
| { |
| LOGGER.info("Creating snapshot on instance snapshotName={} keyspace={} table={} datacenter={} fqdn={}", |
| snapshotName, maybeQuotedKeyspace, maybeQuotedTable, datacenter, ringEntry.fqdn()); |
| SidecarInstance sidecarInstance = new SidecarInstanceImpl(ringEntry.fqdn(), sidecarClientConfig.effectivePort()); |
| ClientConfig.ClearSnapshotStrategy clearSnapshotStrategy = options.clearSnapshotStrategy(); |
| createSnapshotFuture = sidecar.createSnapshot(sidecarInstance, maybeQuotedKeyspace, |
| maybeQuotedTable, snapshotName, clearSnapshotStrategy.ttl()) |
| .handle((resp, throwable) -> { |
| if (throwable == null) |
| { |
| // Create snapshot succeeded |
| return hint; |
| } |
| |
| if (isExhausted(throwable)) |
| { |
| LOGGER.warn("Failed to create snapshot on instance", throwable); |
| return PartitionedDataLayer.AvailabilityHint.DOWN; |
| } |
| |
| LOGGER.error("Unexpected error creating snapshot on instance", throwable); |
| return PartitionedDataLayer.AvailabilityHint.UNKNOWN; |
| }); |
| } |
| |
| return createSnapshotFuture |
| .thenAccept(h -> availabilityHints.put(ringEntry.fqdn(), h)); |
| }) |
| .collect(Collectors.toList()); |
| |
| return CompletableFuture |
| .allOf(futures.toArray(new CompletableFuture[0])) |
| .handle((results, throwable) -> availabilityHints); |
| } |
| |
| protected boolean isExhausted(@Nullable Throwable throwable) |
| { |
| return throwable != null && (throwable instanceof RetriesExhaustedException || isExhausted(throwable.getCause())); |
| } |
| |
| @Override |
| public TimeProvider timeProvider() |
| { |
| return timeProvider; |
| } |
| |
| @Override |
| public boolean useIncrementalRepair() |
| { |
| return useIncrementalRepair; |
| } |
| |
| @Override |
| public boolean readIndexOffset() |
| { |
| return readIndexOffset; |
| } |
| |
| protected void initInstanceMap() |
| { |
| instanceMap = clusterConfig.stream().collect(Collectors.toMap(SidecarInstance::hostname, Function.identity())); |
| try |
| { |
| SslConfigSecretsProvider secretsProvider = sslConfig != null |
| ? new SslConfigSecretsProvider(sslConfig) |
| : null; |
| sidecar = Sidecar.from(new SimpleSidecarInstancesProvider(new ArrayList<>(clusterConfig)), |
| sidecarClientConfig, |
| secretsProvider); |
| } |
| catch (IOException ioException) |
| { |
| throw new RuntimeException("Unable to build sidecar client", ioException); |
| } |
| LOGGER.info("Initialized CassandraDataLayer instanceMap numInstances={}", instanceMap.size()); |
| } |
| |
| @Override |
| public CassandraBridge bridge() |
| { |
| return bridge; |
| } |
| |
| @Override |
| public Stats stats() |
| { |
| return Stats.DoNothingStats.INSTANCE; |
| } |
| |
| @Override |
| public List<SchemaFeature> requestedFeatures() |
| { |
| return requestedFeatures; |
| } |
| |
| @Override |
| public CassandraRing ring() |
| { |
| return tokenPartitioner.ring(); |
| } |
| |
| @Override |
| public TokenPartitioner tokenPartitioner() |
| { |
| return tokenPartitioner; |
| } |
| |
| @Override |
| protected ExecutorService executorService() |
| { |
| return ExecutorHolder.EXECUTOR_SERVICE; |
| } |
| |
| @Override |
| public String jobId() |
| { |
| return null; |
| } |
| |
| @Override |
| public CqlTable cqlTable() |
| { |
| if (cqlTable == null) |
| { |
| throw new RuntimeException("Schema not initialized"); |
| } |
| return cqlTable; |
| } |
| |
| @Override |
| public ReplicationFactor replicationFactor(String keyspace) |
| { |
| return rfMap.get(keyspace); |
| } |
| |
| @Override |
| protected PartitionedDataLayer.AvailabilityHint getAvailability(CassandraInstance instance) |
| { |
| // Hint CassandraInstance availability to parent PartitionedDataLayer |
| PartitionedDataLayer.AvailabilityHint hint = availabilityHints.get(instance.nodeName()); |
| return hint != null ? hint : PartitionedDataLayer.AvailabilityHint.UNKNOWN; |
| } |
| |
| private String snapshotKey(SidecarInstance instance) |
| { |
| return String.format("%s/%s/%d/%s/%s/%s", |
| datacenter, instance.hostname(), instance.port(), keyspace, table, snapshotName); |
| } |
| |
| @Override |
| public CompletableFuture<Stream<SSTable>> listInstance(int partitionId, |
| @NotNull Range<BigInteger> range, |
| @NotNull CassandraInstance instance) |
| { |
| SidecarInstance sidecarInstance = instanceMap.get(instance.nodeName()); |
| if (sidecarInstance == null) |
| { |
| throw new IllegalStateException("Could not find matching cassandra instance: " + instance.nodeName()); |
| } |
| String key = snapshotKey(sidecarInstance); // NOTE: We don't currently support token filtering in list snapshot |
| LOGGER.info("Listing snapshot partition={} lowerBound={} upperBound={} " |
| + "instance={} port={} keyspace={} tableName={} snapshotName={}", |
| partitionId, range.lowerEndpoint(), range.upperEndpoint(), |
| sidecarInstance.hostname(), sidecarInstance.port(), maybeQuotedKeyspace, maybeQuotedTable, snapshotName); |
| try |
| { |
| return SNAPSHOT_CACHE.get(key, () -> { |
| LOGGER.info("Listing instance snapshot partition={} lowerBound={} upperBound={} " |
| + "instance={} port={} keyspace={} tableName={} snapshotName={} cacheKey={}", |
| partitionId, range.lowerEndpoint(), range.upperEndpoint(), |
| sidecarInstance.hostname(), sidecarInstance.port(), maybeQuotedKeyspace, maybeQuotedTable, snapshotName, key); |
| return sidecar.listSnapshotFiles(sidecarInstance, maybeQuotedKeyspace, maybeQuotedTable, snapshotName) |
| .thenApply(response -> collectSSTableList(sidecarInstance, response, partitionId)); |
| }).thenApply(Collection::stream); |
| } |
| catch (ExecutionException exception) |
| { |
| CompletableFuture<Stream<SSTable>> future = new CompletableFuture<>(); |
| future.completeExceptionally(ThrowableUtils.rootCause(exception)); |
| return future; |
| } |
| } |
| |
| @SuppressWarnings("UnstableApiUsage") |
| private List<SSTable> collectSSTableList(SidecarInstance sidecarInstance, |
| ListSnapshotFilesResponse response, |
| int partitionId) |
| { |
| if (response == null) |
| { |
| throw new IncompleteSSTableException(); |
| } |
| List<ListSnapshotFilesResponse.FileInfo> snapshotFilesInfo = response.snapshotFilesInfo(); |
| if (snapshotFilesInfo == null) |
| { |
| throw new IncompleteSSTableException(); |
| } |
| |
| // Group SSTable components together |
| Map<String, Map<FileType, ListSnapshotFilesResponse.FileInfo>> result = new LinkedHashMap<>(1024); |
| for (ListSnapshotFilesResponse.FileInfo file : snapshotFilesInfo) |
| { |
| String fileName = file.fileName; |
| int lastIndexOfDash = fileName.lastIndexOf('-'); |
| if (lastIndexOfDash < 0) |
| { |
| // E.g. dd manifest.json file |
| continue; |
| } |
| String ssTableName = fileName.substring(0, lastIndexOfDash); |
| try |
| { |
| FileType fileType = FileType.fromExtension(fileName.substring(lastIndexOfDash + 1)); |
| result.computeIfAbsent(ssTableName, k -> new LinkedHashMap<>()) |
| .put(fileType, file); |
| } |
| catch (IllegalArgumentException ignore) |
| { |
| // Ignore unknown SSTable component types |
| } |
| } |
| |
| // Map to SSTable |
| return result.values().stream() |
| .map(components -> new SidecarProvisionedSSTable(sidecar, |
| sidecarClientConfig, |
| sidecarInstance, |
| maybeQuotedKeyspace, |
| maybeQuotedTable, |
| snapshotName, |
| components, |
| partitionId, |
| stats())) |
| .collect(Collectors.toList()); |
| } |
| |
| @Override |
| public int hashCode() |
| { |
| return Objects.hash(super.hashCode(), cqlTable, snapshotName, keyspace, table, version()); |
| } |
| |
| @Override |
| public boolean equals(Object other) |
| { |
| if (this == other) |
| { |
| return true; |
| } |
| if (other == null || this.getClass() != other.getClass() || !super.equals(other)) |
| { |
| return false; |
| } |
| |
| CassandraDataLayer that = (CassandraDataLayer) other; |
| return cqlTable.equals(that.cqlTable) |
| && snapshotName.equals(that.snapshotName) |
| && keyspace.equals(that.keyspace) |
| && table.equals(that.table) |
| && version().equals(that.version()); |
| } |
| |
| public Map<String, BigNumberConfigImpl> bigNumberConfigMap() |
| { |
| return bigNumberConfigMap; |
| } |
| |
| @Override |
| public BigNumberConfig bigNumberConfig(CqlField field) |
| { |
| BigNumberConfigImpl config = bigNumberConfigMap.get(field.name()); |
| return config != null ? config : BigNumberConfig.DEFAULT; |
| } |
| |
| /* Internal Cassandra SSTable */ |
| |
| @VisibleForTesting |
| public CassandraRing createCassandraRingFromRing(Partitioner partitioner, |
| ReplicationFactor replicationFactor, |
| RingResponse ring) |
| { |
| Collection<CassandraInstance> instances = ring |
| .stream() |
| .filter(status -> datacenter == null || datacenter.equalsIgnoreCase(status.datacenter())) |
| .map(status -> new CassandraInstance(status.token(), status.fqdn(), status.datacenter())) |
| .collect(Collectors.toList()); |
| return new CassandraRing(partitioner, keyspace, replicationFactor, instances); |
| } |
| |
| // Startup Validation |
| |
| @Override |
| public void startupValidate() |
| { |
| int timeoutSeconds = sidecarClientConfig.timeoutSeconds(); |
| StartupValidator.instance().register(new SidecarValidation(sidecar, timeoutSeconds)); |
| StartupValidator.instance().register(new CassandraValidation(sidecar, timeoutSeconds)); |
| StartupValidator.instance().perform(); |
| } |
| |
| /** |
| * Uses the bridge {@link CassandraBridge#maybeQuoteIdentifier(String)} to attempt to quote the keyspace and table names |
| * when the {@code quoteIdentifiers} option is enabled. |
| */ |
| private void maybeQuoteKeyspaceAndTable() |
| { |
| if (quoteIdentifiers) |
| { |
| Objects.requireNonNull(bridge, "bridge must be initialized to maybe quote keyspace and table"); |
| maybeQuotedKeyspace = bridge.maybeQuoteIdentifier(keyspace); |
| maybeQuotedTable = bridge.maybeQuoteIdentifier(table); |
| } |
| else |
| { |
| // default to unquoted |
| maybeQuotedKeyspace = keyspace; |
| maybeQuotedTable = table; |
| } |
| } |
| |
| // JDK Serialization |
| |
| @SuppressWarnings("unchecked") |
| private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException |
| { |
| LOGGER.debug("Falling back to JDK deserialization"); |
| this.bridge = CassandraBridgeFactory.get(CassandraVersion.valueOf(in.readUTF())); |
| this.snapshotName = in.readUTF(); |
| this.keyspace = readNullable(in); |
| this.table = readNullable(in); |
| this.quoteIdentifiers = in.readBoolean(); |
| this.sidecarClientConfig = Sidecar.ClientConfig.create(in.readInt(), |
| in.readInt(), |
| in.readLong(), |
| in.readLong(), |
| in.readLong(), |
| in.readLong(), |
| in.readInt(), |
| in.readInt(), |
| (Map<FileType, Long>) in.readObject(), |
| (Map<FileType, Long>) in.readObject()); |
| this.sslConfig = (SslConfig) in.readObject(); |
| |
| this.cqlTable = bridge.javaDeserialize(in, CqlTable.class); // Delegate (de-)serialization of version-specific objects to the Cassandra Bridge |
| this.tokenPartitioner = (TokenPartitioner) in.readObject(); |
| this.clusterConfig = (Set<SidecarInstanceImpl>) in.readObject(); |
| this.availabilityHints = (Map<String, AvailabilityHint>) in.readObject(); |
| this.bigNumberConfigMap = (Map<String, BigNumberConfigImpl>) in.readObject(); |
| this.enableStats = in.readBoolean(); |
| this.readIndexOffset = in.readBoolean(); |
| this.useIncrementalRepair = in.readBoolean(); |
| this.lastModifiedTimestampField = readNullable(in); |
| int features = in.readShort(); |
| List<SchemaFeature> requestedFeatures = new ArrayList<>(features); |
| for (int feature = 0; feature < features; feature++) |
| { |
| String featureName = in.readUTF(); |
| requestedFeatures.add(SchemaFeatureSet.valueOf(featureName.toUpperCase())); |
| } |
| this.requestedFeatures = requestedFeatures; |
| // Has alias for last modified timestamp |
| if (this.lastModifiedTimestampField != null) |
| { |
| aliasLastModifiedTimestamp(this.requestedFeatures, this.lastModifiedTimestampField); |
| } |
| this.rfMap = (Map<String, ReplicationFactor>) in.readObject(); |
| this.timeProvider = new ReaderTimeProvider(in.readInt()); |
| this.maybeQuoteKeyspaceAndTable(); |
| this.initInstanceMap(); |
| this.startupValidate(); |
| } |
| |
| private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFoundException |
| { |
| LOGGER.debug("Falling back to JDK serialization"); |
| out.writeUTF(this.version().name()); |
| out.writeUTF(this.snapshotName); |
| writeNullable(out, this.keyspace); |
| writeNullable(out, this.table); |
| out.writeBoolean(this.quoteIdentifiers); |
| out.writeInt(this.sidecarClientConfig.userProvidedPort()); |
| out.writeInt(this.sidecarClientConfig.maxRetries()); |
| out.writeLong(this.sidecarClientConfig.millisToSleep()); |
| out.writeLong(this.sidecarClientConfig.maxMillisToSleep()); |
| out.writeLong(this.sidecarClientConfig.maxBufferSize()); |
| out.writeLong(this.sidecarClientConfig.chunkBufferSize()); |
| out.writeInt(this.sidecarClientConfig.maxPoolSize()); |
| out.writeInt(this.sidecarClientConfig.timeoutSeconds()); |
| out.writeObject(this.sidecarClientConfig.maxBufferOverride()); |
| out.writeObject(this.sidecarClientConfig.chunkBufferOverride()); |
| out.writeObject(this.sslConfig); |
| bridge.javaSerialize(out, this.cqlTable); // Delegate (de-)serialization of version-specific objects to the Cassandra Bridge |
| out.writeObject(this.tokenPartitioner); |
| out.writeObject(this.clusterConfig); |
| out.writeObject(this.availabilityHints); |
| out.writeObject(this.bigNumberConfigMap); |
| out.writeBoolean(this.enableStats); |
| out.writeBoolean(this.readIndexOffset); |
| out.writeBoolean(this.useIncrementalRepair); |
| // If lastModifiedTimestampField exist, it aliases the LMT field |
| writeNullable(out, this.lastModifiedTimestampField); |
| // Write the list of requested features: first write the size, then write the feature names |
| out.writeShort(this.requestedFeatures.size()); |
| for (SchemaFeature feature : requestedFeatures) |
| { |
| out.writeUTF(feature.optionName()); |
| } |
| out.writeObject(this.rfMap); |
| out.writeInt(timeProvider.referenceEpochInSeconds()); |
| } |
| |
| private static void writeNullable(ObjectOutputStream out, @Nullable String string) throws IOException |
| { |
| if (string == null) |
| { |
| out.writeBoolean(false); |
| } |
| else |
| { |
| out.writeBoolean(true); |
| out.writeUTF(string); |
| } |
| } |
| |
| @Nullable |
| private static String readNullable(ObjectInputStream in) throws IOException |
| { |
| if (in.readBoolean()) |
| { |
| return in.readUTF(); |
| } |
| return null; |
| } |
| |
| // Kryo Serialization |
| |
| public static class Serializer extends com.esotericsoftware.kryo.Serializer<CassandraDataLayer> |
| { |
| @Override |
| public void write(Kryo kryo, Output out, CassandraDataLayer dataLayer) |
| { |
| LOGGER.info("Serializing CassandraDataLayer with Kryo"); |
| out.writeString(dataLayer.keyspace); |
| out.writeString(dataLayer.table); |
| out.writeBoolean(dataLayer.quoteIdentifiers); |
| out.writeString(dataLayer.snapshotName); |
| out.writeString(dataLayer.datacenter); |
| out.writeInt(dataLayer.sidecarClientConfig.userProvidedPort()); |
| out.writeInt(dataLayer.sidecarClientConfig.maxRetries()); |
| out.writeLong(dataLayer.sidecarClientConfig.millisToSleep()); |
| out.writeLong(dataLayer.sidecarClientConfig.maxMillisToSleep()); |
| out.writeLong(dataLayer.sidecarClientConfig.maxBufferSize()); |
| out.writeLong(dataLayer.sidecarClientConfig.chunkBufferSize()); |
| out.writeInt(dataLayer.sidecarClientConfig.maxPoolSize()); |
| out.writeInt(dataLayer.sidecarClientConfig.timeoutSeconds()); |
| kryo.writeObject(out, dataLayer.sidecarClientConfig.maxBufferOverride()); |
| kryo.writeObject(out, dataLayer.sidecarClientConfig.chunkBufferOverride()); |
| kryo.writeObjectOrNull(out, dataLayer.sslConfig, SslConfig.class); |
| kryo.writeObject(out, dataLayer.cqlTable); |
| kryo.writeObject(out, dataLayer.tokenPartitioner); |
| kryo.writeObject(out, dataLayer.version()); |
| kryo.writeObject(out, dataLayer.consistencyLevel); |
| kryo.writeObject(out, dataLayer.clusterConfig); |
| kryo.writeObject(out, dataLayer.availabilityHints); |
| out.writeBoolean(dataLayer.bigNumberConfigMap.isEmpty()); // Kryo fails to deserialize bigNumberConfigMap map if empty |
| if (!dataLayer.bigNumberConfigMap.isEmpty()) |
| { |
| kryo.writeObject(out, dataLayer.bigNumberConfigMap); |
| } |
| out.writeBoolean(dataLayer.enableStats); |
| out.writeBoolean(dataLayer.readIndexOffset); |
| out.writeBoolean(dataLayer.useIncrementalRepair); |
| // If lastModifiedTimestampField exist, it aliases the LMT field |
| out.writeString(dataLayer.lastModifiedTimestampField); |
| // Write the list of requested features: first write the size, then write the feature names |
| SchemaFeaturesListWrapper listWrapper = new SchemaFeaturesListWrapper(); |
| listWrapper.requestedFeatureNames = dataLayer.requestedFeatures.stream() |
| .map(SchemaFeature::optionName) |
| .collect(Collectors.toList()); |
| kryo.writeObject(out, listWrapper); |
| kryo.writeObject(out, dataLayer.rfMap); |
| out.writeInt(dataLayer.timeProvider.referenceEpochInSeconds()); |
| } |
| |
| @SuppressWarnings("unchecked") |
| @Override |
| public CassandraDataLayer read(Kryo kryo, Input in, Class<CassandraDataLayer> type) |
| { |
| LOGGER.info("Deserializing CassandraDataLayer with Kryo"); |
| |
| return new CassandraDataLayer( |
| in.readString(), |
| in.readString(), |
| in.readBoolean(), |
| in.readString(), |
| in.readString(), |
| Sidecar.ClientConfig.create(in.readInt(), |
| in.readInt(), |
| in.readLong(), |
| in.readLong(), |
| in.readLong(), |
| in.readLong(), |
| in.readInt(), |
| in.readInt(), |
| (Map<FileType, Long>) kryo.readObject(in, HashMap.class), |
| (Map<FileType, Long>) kryo.readObject(in, HashMap.class)), |
| kryo.readObjectOrNull(in, SslConfig.class), |
| kryo.readObject(in, CqlTable.class), |
| kryo.readObject(in, TokenPartitioner.class), |
| kryo.readObject(in, CassandraVersion.class), |
| kryo.readObject(in, ConsistencyLevel.class), |
| kryo.readObject(in, HashSet.class), |
| (Map<String, PartitionedDataLayer.AvailabilityHint>) kryo.readObject(in, HashMap.class), |
| in.readBoolean() ? Collections.emptyMap() |
| : (Map<String, BigNumberConfigImpl>) kryo.readObject(in, HashMap.class), |
| in.readBoolean(), |
| in.readBoolean(), |
| in.readBoolean(), |
| in.readString(), |
| kryo.readObject(in, SchemaFeaturesListWrapper.class).toList(), |
| kryo.readObject(in, HashMap.class), |
| new ReaderTimeProvider(in.readInt())); |
| } |
| |
| // Wrapper only used internally for Kryo serialization/deserialization |
| private static class SchemaFeaturesListWrapper |
| { |
| public List<String> requestedFeatureNames; // CHECKSTYLE IGNORE: Public mutable field |
| |
| public List<SchemaFeature> toList() |
| { |
| return requestedFeatureNames.stream() |
| .map(name -> SchemaFeatureSet.valueOf(name.toUpperCase())) |
| .collect(Collectors.toList()); |
| } |
| } |
| } |
| |
| protected Set<? extends SidecarInstance> initializeClusterConfig(ClientConfig options) |
| { |
| return Arrays.stream(options.sidecarInstances().split(",")) |
| .map(hostname -> new SidecarInstanceImpl(hostname, options.sidecarPort())) |
| .collect(Collectors.toSet()); |
| } |
| |
| protected String getEffectiveCassandraVersionForRead(Set<? extends SidecarInstance> clusterConfig, |
| NodeSettings nodeSettings) |
| { |
| return nodeSettings.releaseVersion(); |
| } |
| |
| protected void dialHome(@NotNull ClientConfig options) |
| { |
| LOGGER.info("Dial home. clientConfig={}", options); |
| } |
| |
| protected void clearSnapshot(Set<? extends SidecarInstance> clusterConfig, @NotNull ClientConfig options) |
| { |
| if (maybeQuotedKeyspace == null || maybeQuotedTable == null) |
| { |
| LOGGER.warn("Bridge was never initialized. Skipping clearing snapshots. This implies snapshots were never created"); |
| return; |
| } |
| |
| LOGGER.info("Clearing snapshot at end of Spark job snapshotName={} keyspace={} table={} dc={}", |
| snapshotName, maybeQuotedKeyspace, maybeQuotedTable, datacenter); |
| CountDownLatch latch = new CountDownLatch(clusterConfig.size()); |
| try |
| { |
| for (SidecarInstance instance : clusterConfig) |
| { |
| sidecar.clearSnapshot(instance, maybeQuotedKeyspace, maybeQuotedTable, snapshotName).whenComplete((resp, throwable) -> { |
| try |
| { |
| if (throwable != null) |
| { |
| LOGGER.warn("Failed to clear snapshot on instance hostname={} port={} snapshotName={} keyspace={} table={} datacenter={}", |
| instance.hostname(), instance.port(), snapshotName, maybeQuotedKeyspace, maybeQuotedTable, datacenter, throwable); |
| } |
| } |
| finally |
| { |
| latch.countDown(); |
| } |
| }); |
| } |
| await(latch); |
| LOGGER.info("Snapshot cleared snapshotName={} keyspace={} table={} datacenter={}", |
| snapshotName, maybeQuotedKeyspace, maybeQuotedTable, datacenter); |
| } |
| catch (Throwable throwable) |
| { |
| LOGGER.warn("Unexpected exception clearing snapshot snapshotName={} keyspace={} table={} dc={}", |
| snapshotName, maybeQuotedKeyspace, maybeQuotedTable, datacenter, throwable); |
| } |
| } |
| |
| /** |
| * Returns the {@link Sizing} object based on the {@code sizing} option provided by the user, |
| * or {@link DefaultSizing} as the default sizing |
| * |
| * @param clusterConfig the cluster configuration |
| * @param replicationFactor the replication factor |
| * @param options the {@link ClientConfig} options |
| * @return the {@link Sizing} object based on the {@code sizing} option provided by the user |
| */ |
| protected Sizing getSizing(Set<? extends SidecarInstance> clusterConfig, |
| ReplicationFactor replicationFactor, |
| ClientConfig options) |
| { |
| return new DefaultSizing(options.numCores()); |
| } |
| |
| protected void await(CountDownLatch latch) |
| { |
| try |
| { |
| latch.await(); |
| } |
| catch (InterruptedException exception) |
| { |
| Thread.currentThread().interrupt(); |
| throw new RuntimeException(exception); |
| } |
| } |
| |
| static void aliasLastModifiedTimestamp(List<SchemaFeature> requestedFeatures, String alias) |
| { |
| SchemaFeature featureAlias = new SchemaFeature() |
| { |
| @Override |
| public String optionName() |
| { |
| return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.optionName(); |
| } |
| |
| @Override |
| public String fieldName() |
| { |
| return alias; |
| } |
| |
| @Override |
| public DataType fieldDataType() |
| { |
| return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.fieldDataType(); |
| } |
| |
| @Override |
| public RowBuilder decorate(RowBuilder builder) |
| { |
| return new LastModifiedTimestampDecorator(builder, alias); |
| } |
| |
| @Override |
| public boolean fieldNullable() |
| { |
| return SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP.fieldNullable(); |
| } |
| }; |
| int index = requestedFeatures.indexOf(SchemaFeatureSet.LAST_MODIFIED_TIMESTAMP); |
| if (index >= 0) |
| { |
| requestedFeatures.set(index, featureAlias); |
| } |
| } |
| } |
