http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java new file mode 100644 index 0000000..d10b9c5 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java @@ -0,0 +1,703 @@ +/* + * 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.repair.consistent; + +import java.io.IOException; +import java.net.InetAddress; +import java.nio.ByteBuffer; +import java.time.Instant; +import java.util.Date; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import javax.annotation.Nullable; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Iterables; +import com.google.common.collect.Lists; +import com.google.common.primitives.Ints; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.db.marshal.InetAddressType; +import org.apache.cassandra.db.marshal.UUIDType; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.io.util.DataInputBuffer; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.net.MessageOut; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.repair.messages.FailSession; +import org.apache.cassandra.repair.messages.FinalizeCommit; +import org.apache.cassandra.repair.messages.FinalizePromise; +import org.apache.cassandra.repair.messages.FinalizePropose; +import org.apache.cassandra.repair.messages.PrepareConsistentRequest; +import org.apache.cassandra.repair.messages.PrepareConsistentResponse; +import org.apache.cassandra.repair.messages.RepairMessage; +import org.apache.cassandra.repair.messages.StatusRequest; +import org.apache.cassandra.repair.messages.StatusResponse; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.FBUtilities; + +import static org.apache.cassandra.repair.consistent.ConsistentSession.State.*; + +/** + * Manages all consistent repair sessions a node is participating in. + * <p/> + * Since sessions need to be loaded, and since we need to handle cases where sessions might not exist, most of the logic + * around local sessions is implemented in this class, with the LocalSession class being treated more like a simple struct, + * in contrast with {@link CoordinatorSession} + */ +public class LocalSessions +{ + private static final Logger logger = LoggerFactory.getLogger(LocalSessions.class); + + /** + * Amount of time a session can go without any activity before we start checking the status of other + * participants to see if we've missed a message + */ + static final int CHECK_STATUS_TIMEOUT = Integer.getInteger("cassandra.repair_status_check_timeout_seconds", + Ints.checkedCast(TimeUnit.HOURS.toSeconds(1))); + + /** + * Amount of time a session can go without any activity before being automatically set to FAILED + */ + static final int AUTO_FAIL_TIMEOUT = Integer.getInteger("cassandra.repair_fail_timeout_seconds", + Ints.checkedCast(TimeUnit.DAYS.toSeconds(1))); + + /** + * Amount of time a completed session is kept around after completion before being deleted, this gives + * compaction plenty of time to move sstables from successful sessions into the repaired bucket + */ + static final int AUTO_DELETE_TIMEOUT = Integer.getInteger("cassandra.repair_delete_timeout_seconds", + Ints.checkedCast(TimeUnit.DAYS.toSeconds(1))); + /** + * How often LocalSessions.cleanup is run + */ + public static final int CLEANUP_INTERVAL = Integer.getInteger("cassandra.repair_cleanup_interval_seconds", + Ints.checkedCast(TimeUnit.MINUTES.toSeconds(10))); + + private static Set<TableId> uuidToTableId(Set<UUID> src) + { + return ImmutableSet.copyOf(Iterables.transform(src, TableId::fromUUID)); + } + + private static Set<UUID> tableIdToUuid(Set<TableId> src) + { + return ImmutableSet.copyOf(Iterables.transform(src, TableId::asUUID)); + } + + private final String keyspace = SchemaConstants.SYSTEM_KEYSPACE_NAME; + private final String table = SystemKeyspace.REPAIRS; + private boolean started = false; + private volatile ImmutableMap<UUID, LocalSession> sessions = ImmutableMap.of(); + + @VisibleForTesting + int getNumSessions() + { + return sessions.size(); + } + + @VisibleForTesting + protected InetAddress getBroadcastAddress() + { + return FBUtilities.getBroadcastAddress(); + } + + @VisibleForTesting + protected boolean isAlive(InetAddress address) + { + return FailureDetector.instance.isAlive(address); + } + + @VisibleForTesting + protected boolean isNodeInitialized() + { + return StorageService.instance.isInitialized(); + } + + public List<Map<String, String>> sessionInfo(boolean all) + { + Iterable<LocalSession> currentSessions = sessions.values(); + if (!all) + { + currentSessions = Iterables.filter(currentSessions, s -> !s.isCompleted()); + } + return Lists.newArrayList(Iterables.transform(currentSessions, LocalSessionInfo::sessionToMap)); + } + + /** + * hook for operators to cancel sessions, cancelling from a non-coordinator is an error, unless + * force is set to true. Messages are sent out to other participants, but we don't wait for a response + */ + public void cancelSession(UUID sessionID, boolean force) + { + logger.debug("cancelling session {}", sessionID); + LocalSession session = getSession(sessionID); + Preconditions.checkArgument(session != null, "Session {} does not exist", sessionID); + Preconditions.checkArgument(force || session.coordinator.equals(getBroadcastAddress()), + "Cancel session %s from it's coordinator (%s) or use --force", + sessionID, session.coordinator); + + setStateAndSave(session, FAILED); + for (InetAddress participant : session.participants) + { + if (!participant.equals(getBroadcastAddress())) + sendMessage(participant, new FailSession(sessionID)); + } + } + + /** + * Loads sessions out of the repairs table and sets state to started + */ + public synchronized void start() + { + Preconditions.checkArgument(!started, "LocalSessions.start can only be called once"); + Preconditions.checkArgument(sessions.isEmpty(), "No sessions should be added before start"); + UntypedResultSet rows = QueryProcessor.executeInternalWithPaging(String.format("SELECT * FROM %s.%s", keyspace, table), 1000); + Map<UUID, LocalSession> loadedSessions = new HashMap<>(); + for (UntypedResultSet.Row row : rows) + { + try + { + LocalSession session = load(row); + loadedSessions.put(session.sessionID, session); + } + catch (IllegalArgumentException | NullPointerException e) + { + logger.warn("Unable to load malformed repair session {}, ignoring", row.has("parent_id") ? row.getUUID("parent_id") : null); + } + } + sessions = ImmutableMap.copyOf(loadedSessions); + started = true; + } + + public boolean isStarted() + { + return started; + } + + private static boolean shouldCheckStatus(LocalSession session, int now) + { + return !session.isCompleted() && (now > session.getLastUpdate() + CHECK_STATUS_TIMEOUT); + } + + private static boolean shouldFail(LocalSession session, int now) + { + return !session.isCompleted() && (now > session.getLastUpdate() + AUTO_FAIL_TIMEOUT); + } + + private static boolean shouldDelete(LocalSession session, int now) + { + return session.isCompleted() && (now > session.getLastUpdate() + AUTO_DELETE_TIMEOUT); + } + + /** + * Auto fails and auto deletes timed out and old sessions + * Compaction will clean up the sstables still owned by a deleted session + */ + public void cleanup() + { + logger.debug("Running LocalSessions.cleanup"); + if (!isNodeInitialized()) + { + logger.debug("node not initialized, aborting local session cleanup"); + return; + } + Set<LocalSession> currentSessions = new HashSet<>(sessions.values()); + for (LocalSession session : currentSessions) + { + synchronized (session) + { + int now = FBUtilities.nowInSeconds(); + if (shouldFail(session, now)) + { + logger.warn("Auto failing timed out repair session {}", session); + failSession(session.sessionID, false); + } + else if (shouldDelete(session, now)) + { + logger.warn("Auto deleting repair session {}", session); + deleteSession(session.sessionID); + } + else if (shouldCheckStatus(session, now)) + { + sendStatusRequest(session); + } + } + } + } + + private static ByteBuffer serializeRange(Range<Token> range) + { + int size = (int) Token.serializer.serializedSize(range.left, 0); + size += (int) Token.serializer.serializedSize(range.right, 0); + try (DataOutputBuffer buffer = new DataOutputBuffer(size)) + { + Token.serializer.serialize(range.left, buffer, 0); + Token.serializer.serialize(range.right, buffer, 0); + return buffer.buffer(); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + private static Set<ByteBuffer> serializeRanges(Set<Range<Token>> ranges) + { + Set<ByteBuffer> buffers = new HashSet<>(ranges.size()); + ranges.forEach(r -> buffers.add(serializeRange(r))); + return buffers; + } + + private static Range<Token> deserializeRange(ByteBuffer bb) + { + try (DataInputBuffer in = new DataInputBuffer(bb, false)) + { + IPartitioner partitioner = DatabaseDescriptor.getPartitioner(); + Token left = Token.serializer.deserialize(in, partitioner, 0); + Token right = Token.serializer.deserialize(in, partitioner, 0); + return new Range<>(left, right); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + private static Set<Range<Token>> deserializeRanges(Set<ByteBuffer> buffers) + { + Set<Range<Token>> ranges = new HashSet<>(buffers.size()); + buffers.forEach(bb -> ranges.add(deserializeRange(bb))); + return ranges; + } + + /** + * Save session state to table + */ + @VisibleForTesting + void save(LocalSession session) + { + String query = "INSERT INTO %s.%s " + + "(parent_id, " + + "started_at, " + + "last_update, " + + "repaired_at, " + + "state, " + + "coordinator, " + + "participants, " + + "ranges, " + + "cfids) " + + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"; + + QueryProcessor.executeInternal(String.format(query, keyspace, table), + session.sessionID, + Date.from(Instant.ofEpochSecond(session.startedAt)), + Date.from(Instant.ofEpochSecond(session.getLastUpdate())), + Date.from(Instant.ofEpochMilli(session.repairedAt)), + session.getState().ordinal(), + session.coordinator, + session.participants, + serializeRanges(session.ranges), + tableIdToUuid(session.tableIds)); + } + + private static int dateToSeconds(Date d) + { + return Ints.checkedCast(TimeUnit.MILLISECONDS.toSeconds(d.getTime())); + } + + private LocalSession load(UntypedResultSet.Row row) + { + LocalSession.Builder builder = LocalSession.builder(); + builder.withState(ConsistentSession.State.valueOf(row.getInt("state"))); + builder.withSessionID(row.getUUID("parent_id")); + builder.withCoordinator(row.getInetAddress("coordinator")); + builder.withTableIds(uuidToTableId(row.getSet("cfids", UUIDType.instance))); + builder.withRepairedAt(row.getTimestamp("repaired_at").getTime()); + builder.withRanges(deserializeRanges(row.getSet("ranges", BytesType.instance))); + builder.withParticipants(row.getSet("participants", InetAddressType.instance)); + + builder.withStartedAt(dateToSeconds(row.getTimestamp("started_at"))); + builder.withLastUpdate(dateToSeconds(row.getTimestamp("last_update"))); + + return buildSession(builder); + } + + private void deleteRow(UUID sessionID) + { + String query = "DELETE FROM %s.%s WHERE parent_id=?"; + QueryProcessor.executeInternal(String.format(query, keyspace, table), sessionID); + } + + /** + * Loads a session directly from the table. Should be used for testing only + */ + @VisibleForTesting + LocalSession loadUnsafe(UUID sessionId) + { + String query = "SELECT * FROM %s.%s WHERE parent_id=?"; + UntypedResultSet result = QueryProcessor.executeInternal(String.format(query, keyspace, table), sessionId); + if (result.isEmpty()) + return null; + + UntypedResultSet.Row row = result.one(); + return load(row); + } + + @VisibleForTesting + protected LocalSession buildSession(LocalSession.Builder builder) + { + return new LocalSession(builder); + } + + protected LocalSession getSession(UUID sessionID) + { + return sessions.get(sessionID); + } + + @VisibleForTesting + synchronized void putSessionUnsafe(LocalSession session) + { + putSession(session); + save(session); + } + + private synchronized void putSession(LocalSession session) + { + Preconditions.checkArgument(!sessions.containsKey(session.sessionID), + "LocalSession {} already exists", session.sessionID); + Preconditions.checkArgument(started, "sessions cannot be added before LocalSessions is started"); + sessions = ImmutableMap.<UUID, LocalSession>builder() + .putAll(sessions) + .put(session.sessionID, session) + .build(); + } + + private synchronized void removeSession(UUID sessionID) + { + Preconditions.checkArgument(sessionID != null); + Map<UUID, LocalSession> temp = new HashMap<>(sessions); + temp.remove(sessionID); + sessions = ImmutableMap.copyOf(temp); + } + + @VisibleForTesting + LocalSession createSessionUnsafe(UUID sessionId, ActiveRepairService.ParentRepairSession prs, Set<InetAddress> peers) + { + LocalSession.Builder builder = LocalSession.builder(); + builder.withState(ConsistentSession.State.PREPARING); + builder.withSessionID(sessionId); + builder.withCoordinator(prs.coordinator); + + builder.withTableIds(prs.getTableIds()); + builder.withRepairedAt(prs.repairedAt); + builder.withRanges(prs.getRanges()); + builder.withParticipants(peers); + + int now = FBUtilities.nowInSeconds(); + builder.withStartedAt(now); + builder.withLastUpdate(now); + + return buildSession(builder); + } + + protected ActiveRepairService.ParentRepairSession getParentRepairSession(UUID sessionID) + { + return ActiveRepairService.instance.getParentRepairSession(sessionID); + } + + protected void sendMessage(InetAddress destination, RepairMessage message) + { + logger.debug("sending {} to {}", message, destination); + MessageOut<RepairMessage> messageOut = new MessageOut<RepairMessage>(MessagingService.Verb.REPAIR_MESSAGE, message, RepairMessage.serializer); + MessagingService.instance().sendOneWay(messageOut, destination); + } + + private void setStateAndSave(LocalSession session, ConsistentSession.State state) + { + synchronized (session) + { + Preconditions.checkArgument(session.getState().canTransitionTo(state), + "Invalid state transition %s -> %s", + session.getState(), state); + logger.debug("Setting LocalSession state from {} -> {} for {}", session.getState(), state, session.sessionID); + session.setState(state); + session.setLastUpdate(); + save(session); + } + } + + public void failSession(UUID sessionID) + { + failSession(sessionID, true); + } + + public void failSession(UUID sessionID, boolean sendMessage) + { + logger.debug("failing session {}", sessionID); + LocalSession session = getSession(sessionID); + if (session != null) + { + setStateAndSave(session, FAILED); + if (sendMessage) + { + sendMessage(session.coordinator, new FailSession(sessionID)); + } + } + } + + public synchronized void deleteSession(UUID sessionID) + { + logger.debug("deleting session {}", sessionID); + LocalSession session = getSession(sessionID); + Preconditions.checkArgument(session.isCompleted(), "Cannot delete incomplete sessions"); + + deleteRow(sessionID); + removeSession(sessionID); + } + + @VisibleForTesting + ListenableFuture submitPendingAntiCompaction(LocalSession session, ExecutorService executor) + { + PendingAntiCompaction pac = new PendingAntiCompaction(session.sessionID, session.ranges, executor); + return pac.run(); + } + + /** + * The PrepareConsistentRequest effectively promotes the parent repair session to a consistent + * incremental session, and begins the 'pending anti compaction' which moves all sstable data + * that is to be repaired into it's own silo, preventing it from mixing with other data. + * + * No response is sent to the repair coordinator until the pending anti compaction has completed + * successfully. If the pending anti compaction fails, a failure message is sent to the coordinator, + * cancelling the session. + */ + public void handlePrepareMessage(InetAddress from, PrepareConsistentRequest request) + { + logger.debug("received {} from {}", request, from); + UUID sessionID = request.parentSession; + InetAddress coordinator = request.coordinator; + Set<InetAddress> peers = request.participants; + + ActiveRepairService.ParentRepairSession parentSession; + try + { + parentSession = getParentRepairSession(sessionID); + } + catch (Throwable e) + { + logger.debug("Error retrieving ParentRepairSession for session {}, responding with failure", sessionID); + sendMessage(coordinator, new FailSession(sessionID)); + return; + } + + LocalSession session = createSessionUnsafe(sessionID, parentSession, peers); + putSessionUnsafe(session); + logger.debug("created local session for {}", sessionID); + + ExecutorService executor = Executors.newFixedThreadPool(parentSession.getColumnFamilyStores().size()); + + ListenableFuture pendingAntiCompaction = submitPendingAntiCompaction(session, executor); + Futures.addCallback(pendingAntiCompaction, new FutureCallback() + { + public void onSuccess(@Nullable Object result) + { + logger.debug("pending anti-compaction for {} completed", sessionID); + setStateAndSave(session, PREPARED); + sendMessage(coordinator, new PrepareConsistentResponse(sessionID, getBroadcastAddress(), true)); + executor.shutdown(); + } + + public void onFailure(Throwable t) + { + logger.debug("pending anti-compaction for {} failed", sessionID); + failSession(sessionID); + executor.shutdown(); + } + }); + } + + public void maybeSetRepairing(UUID sessionID) + { + LocalSession session = getSession(sessionID); + if (session != null && session.getState() != REPAIRING) + { + logger.debug("Setting local session {} to REPAIRING", session); + setStateAndSave(session, REPAIRING); + } + } + + public void handleFinalizeProposeMessage(InetAddress from, FinalizePropose propose) + { + logger.debug("received {} from {}", propose, from); + UUID sessionID = propose.sessionID; + LocalSession session = getSession(sessionID); + if (session == null) + { + logger.debug("No LocalSession found for session {}, responding with failure", sessionID); + sendMessage(from, new FailSession(sessionID)); + return; + } + + try + { + setStateAndSave(session, FINALIZE_PROMISED); + sendMessage(from, new FinalizePromise(sessionID, getBroadcastAddress(), true)); + } + catch (IllegalArgumentException e) + { + logger.error("error setting session to FINALIZE_PROMISED", e); + failSession(sessionID); + } + } + + /** + * Finalizes the repair session, completing it as successful. + * + * This only changes the state of the session, it doesn't promote the siloed sstables to repaired. That will happen + * as part of the compaction process, and avoids having to worry about in progress compactions interfering with the + * promotion. + */ + public void handleFinalizeCommitMessage(InetAddress from, FinalizeCommit commit) + { + logger.debug("received {} from {}", commit, from); + UUID sessionID = commit.sessionID; + LocalSession session = getSession(sessionID); + if (session == null) + { + logger.warn("Received finalize commit message for unknown session {}", sessionID); + return; + } + + setStateAndSave(session, FINALIZED); + } + + public void handleFailSessionMessage(InetAddress from, FailSession msg) + { + logger.debug("received {} from {}", msg, from); + failSession(msg.sessionID, false); + } + + public void sendStatusRequest(LocalSession session) + { + StatusRequest request = new StatusRequest(session.sessionID); + for (InetAddress participant : session.participants) + { + if (!getBroadcastAddress().equals(participant) && isAlive(participant)) + { + sendMessage(participant, request); + } + } + } + + public void handleStatusRequest(InetAddress from, StatusRequest request) + { + logger.debug("received {} from {}", request, from); + UUID sessionID = request.sessionID; + LocalSession session = getSession(sessionID); + if (session == null) + { + logger.warn("Received status response message for unknown session {}", sessionID); + sendMessage(from, new StatusResponse(sessionID, FAILED)); + } + else + { + sendMessage(from, new StatusResponse(sessionID, session.getState())); + } + } + + public void handleStatusResponse(InetAddress from, StatusResponse response) + { + logger.debug("received {} from {}", response, from); + UUID sessionID = response.sessionID; + LocalSession session = getSession(sessionID); + if (session == null) + { + logger.warn("Received status response message for unknown session {}", sessionID); + return; + } + + // only change local state if response state is FINALIZED or FAILED, since those are + // the only statuses that would indicate we've missed a message completing the session + if (response.state == FINALIZED || response.state == FAILED) + { + setStateAndSave(session, response.state); + } + else + { + logger.debug("{} is not actionable"); + } + } + + /** + * determines if a local session exists, and if it's not finalized or failed + */ + public boolean isSessionInProgress(UUID sessionID) + { + LocalSession session = getSession(sessionID); + return session != null && session.getState() != FINALIZED && session.getState() != FAILED; + } + + /** + * Returns the repairedAt time for a sessions which is unknown, failed, or finalized + * calling this for a session which is in progress throws an exception + */ + public long getFinalSessionRepairedAt(UUID sessionID) + { + LocalSession session = getSession(sessionID); + if (session == null || session.getState() == FAILED) + { + return ActiveRepairService.UNREPAIRED_SSTABLE; + } + else if (session.getState() == FINALIZED) + { + return session.repairedAt; + } + else + { + throw new IllegalStateException("Cannot get final repaired at value for in progress session: " + session); + } + } +}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/consistent/PendingAntiCompaction.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/consistent/PendingAntiCompaction.java b/src/java/org/apache/cassandra/repair/consistent/PendingAntiCompaction.java new file mode 100644 index 0000000..5203c41 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/consistent/PendingAntiCompaction.java @@ -0,0 +1,195 @@ +/* + * 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.repair.consistent; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; + +import com.google.common.collect.Iterables; +import com.google.common.collect.Lists; +import com.google.common.util.concurrent.AsyncFunction; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListenableFutureTask; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.db.compaction.OperationType; +import org.apache.cassandra.db.lifecycle.LifecycleTransaction; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.utils.concurrent.Refs; + +/** + * Performs an anti compaction on a set of tables and token ranges, isolating the unrepaired sstables + * for a give token range into a pending repair group so they can't be compacted with other sstables + * while they are being repaired. + */ +public class PendingAntiCompaction +{ + private static final Logger logger = LoggerFactory.getLogger(PendingAntiCompaction.class); + + static class AcquireResult + { + final ColumnFamilyStore cfs; + final Refs<SSTableReader> refs; + final LifecycleTransaction txn; + + AcquireResult(ColumnFamilyStore cfs, Refs<SSTableReader> refs, LifecycleTransaction txn) + { + this.cfs = cfs; + this.refs = refs; + this.txn = txn; + } + + void abort() + { + txn.abort(); + refs.release(); + } + } + + static class AcquisitionCallable implements Callable<AcquireResult> + { + private final ColumnFamilyStore cfs; + private final Collection<Range<Token>> ranges; + private final UUID sessionID; + + public AcquisitionCallable(ColumnFamilyStore cfs, Collection<Range<Token>> ranges, UUID sessionID) + { + this.cfs = cfs; + this.ranges = ranges; + this.sessionID = sessionID; + } + + private Iterable<SSTableReader> getSSTables() + { + return Iterables.filter(cfs.getLiveSSTables(), s -> !s.isRepaired() && !s.isPendingRepair() && s.intersects(ranges)); + } + + private AcquireResult acquireTuple() + { + List<SSTableReader> sstables = Lists.newArrayList(getSSTables()); + if (sstables.isEmpty()) + return new AcquireResult(cfs, null, null); + + LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION); + if (txn != null) + return new AcquireResult(cfs, Refs.ref(sstables), txn); + else + return null; + } + + public AcquireResult call() throws Exception + { + logger.debug("acquiring sstables for pending anti compaction on session {}", sessionID); + AcquireResult refTxn = acquireTuple(); + if (refTxn != null) + return refTxn; + + // try to modify after cancelling running compactions. This will attempt to cancel in flight compactions for + // up to a minute, after which point, null will be returned + return cfs.runWithCompactionsDisabled(this::acquireTuple, false, false); + } + } + + static class AcquisitionCallback implements AsyncFunction<List<AcquireResult>, Object> + { + private final UUID parentRepairSession; + private final Collection<Range<Token>> ranges; + + public AcquisitionCallback(UUID parentRepairSession, Collection<Range<Token>> ranges) + { + this.parentRepairSession = parentRepairSession; + this.ranges = ranges; + } + + ListenableFuture<?> submitPendingAntiCompaction(AcquireResult result) + { + return CompactionManager.instance.submitPendingAntiCompaction(result.cfs, ranges, result.refs, result.txn, parentRepairSession); + } + + public ListenableFuture apply(List<AcquireResult> results) throws Exception + { + if (Iterables.any(results, t -> t == null)) + { + // Release all sstables, and report failure back to coordinator + for (AcquireResult result : results) + { + if (result != null) + { + logger.info("Releasing acquired sstables for {}.{}", result.cfs.metadata.keyspace, result.cfs.metadata.name); + result.abort(); + } + } + return Futures.immediateFailedFuture(new RuntimeException("unable to acquire sstables")); + } + else + { + List<ListenableFuture<?>> pendingAntiCompactions = new ArrayList<>(results.size()); + for (AcquireResult result : results) + { + if (result.txn != null) + { + ListenableFuture<?> future = submitPendingAntiCompaction(result); + pendingAntiCompactions.add(future); + } + } + + return Futures.allAsList(pendingAntiCompactions); + } + } + } + + private final UUID prsId; + private final Collection<Range<Token>> ranges; + private final ExecutorService executor; + + public PendingAntiCompaction(UUID prsId, Collection<Range<Token>> ranges, ExecutorService executor) + { + this.prsId = prsId; + this.ranges = ranges; + this.executor = executor; + } + + public ListenableFuture run() + { + ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(prsId); + List<ListenableFutureTask<AcquireResult>> tasks = new ArrayList<>(); + for (ColumnFamilyStore cfs : prs.getColumnFamilyStores()) + { + cfs.forceBlockingFlush(); + ListenableFutureTask<AcquireResult> task = ListenableFutureTask.create(new AcquisitionCallable(cfs, ranges, prsId)); + executor.submit(task); + tasks.add(task); + } + ListenableFuture<List<AcquireResult>> acquisitionResults = Futures.successfulAsList(tasks); + ListenableFuture compactionResult = Futures.transform(acquisitionResults, new AcquisitionCallback(prsId, ranges)); + return compactionResult; + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/AnticompactionRequest.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/AnticompactionRequest.java b/src/java/org/apache/cassandra/repair/messages/AnticompactionRequest.java deleted file mode 100644 index a29cc87..0000000 --- a/src/java/org/apache/cassandra/repair/messages/AnticompactionRequest.java +++ /dev/null @@ -1,107 +0,0 @@ -/* - * 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.repair.messages; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; -import java.util.Objects; -import java.util.UUID; - -import org.apache.cassandra.dht.Range; -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.io.util.DataInputPlus; -import org.apache.cassandra.io.util.DataOutputPlus; -import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.utils.UUIDSerializer; - -public class AnticompactionRequest extends RepairMessage -{ - public static MessageSerializer serializer = new AnticompactionRequestSerializer(); - public final UUID parentRepairSession; - /** - * Successfully repaired ranges. Does not contain null. - */ - public final Collection<Range<Token>> successfulRanges; - - public AnticompactionRequest(UUID parentRepairSession, Collection<Range<Token>> ranges) - { - super(Type.ANTICOMPACTION_REQUEST, null); - this.parentRepairSession = parentRepairSession; - this.successfulRanges = ranges; - } - - @Override - public boolean equals(Object o) - { - if (!(o instanceof AnticompactionRequest)) - return false; - AnticompactionRequest other = (AnticompactionRequest)o; - return messageType == other.messageType && - parentRepairSession.equals(other.parentRepairSession) && - successfulRanges.equals(other.successfulRanges); - } - - @Override - public int hashCode() - { - return Objects.hash(messageType, parentRepairSession, successfulRanges); - } - - public static class AnticompactionRequestSerializer implements MessageSerializer<AnticompactionRequest> - { - public void serialize(AnticompactionRequest message, DataOutputPlus out, int version) throws IOException - { - UUIDSerializer.serializer.serialize(message.parentRepairSession, out, version); - out.writeInt(message.successfulRanges.size()); - for (Range<Token> r : message.successfulRanges) - { - MessagingService.validatePartitioner(r); - Range.tokenSerializer.serialize(r, out, version); - } - } - - public AnticompactionRequest deserialize(DataInputPlus in, int version) throws IOException - { - UUID parentRepairSession = UUIDSerializer.serializer.deserialize(in, version); - int rangeCount = in.readInt(); - List<Range<Token>> ranges = new ArrayList<>(rangeCount); - for (int i = 0; i < rangeCount; i++) - ranges.add((Range<Token>) Range.tokenSerializer.deserialize(in, MessagingService.globalPartitioner(), version)); - return new AnticompactionRequest(parentRepairSession, ranges); - } - - public long serializedSize(AnticompactionRequest message, int version) - { - long size = UUIDSerializer.serializer.serializedSize(message.parentRepairSession, version); - size += Integer.BYTES; // count of items in successfulRanges - for (Range<Token> r : message.successfulRanges) - size += Range.tokenSerializer.serializedSize(r, version); - return size; - } - } - - @Override - public String toString() - { - return "AnticompactionRequest{" + - "parentRepairSession=" + parentRepairSession + - "} " + super.toString(); - } -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/FailSession.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/FailSession.java b/src/java/org/apache/cassandra/repair/messages/FailSession.java new file mode 100644 index 0000000..1227cc3 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/FailSession.java @@ -0,0 +1,71 @@ +/* + * 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.repair.messages; + +import java.io.IOException; +import java.util.UUID; + +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.utils.UUIDSerializer; + +public class FailSession extends RepairMessage +{ + public final UUID sessionID; + + public FailSession(UUID sessionID) + { + super(Type.FAILED_SESSION, null); + assert sessionID != null; + this.sessionID = sessionID; + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + FailSession that = (FailSession) o; + + return sessionID.equals(that.sessionID); + } + + public int hashCode() + { + return sessionID.hashCode(); + } + + public static final MessageSerializer serializer = new MessageSerializer<FailSession>() + { + public void serialize(FailSession msg, DataOutputPlus out, int version) throws IOException + { + UUIDSerializer.serializer.serialize(msg.sessionID, out, version); + } + + public FailSession deserialize(DataInputPlus in, int version) throws IOException + { + return new FailSession(UUIDSerializer.serializer.deserialize(in, version)); + } + + public long serializedSize(FailSession msg, int version) + { + return UUIDSerializer.serializer.serializedSize(msg.sessionID, version); + } + }; +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java b/src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java new file mode 100644 index 0000000..a4eb111 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java @@ -0,0 +1,78 @@ +/* + * 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.repair.messages; + +import java.io.IOException; +import java.util.UUID; + +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.utils.UUIDSerializer; + +public class FinalizeCommit extends RepairMessage +{ + public final UUID sessionID; + + public FinalizeCommit(UUID sessionID) + { + super(Type.FINALIZE_COMMIT, null); + assert sessionID != null; + this.sessionID = sessionID; + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + FinalizeCommit that = (FinalizeCommit) o; + + return sessionID.equals(that.sessionID); + } + + public int hashCode() + { + return sessionID.hashCode(); + } + + public String toString() + { + return "FinalizeCommit{" + + "sessionID=" + sessionID + + '}'; + } + + public static MessageSerializer serializer = new MessageSerializer<FinalizeCommit>() + { + public void serialize(FinalizeCommit msg, DataOutputPlus out, int version) throws IOException + { + UUIDSerializer.serializer.serialize(msg.sessionID, out, version); + } + + public FinalizeCommit deserialize(DataInputPlus in, int version) throws IOException + { + return new FinalizeCommit(UUIDSerializer.serializer.deserialize(in, version)); + } + + public long serializedSize(FinalizeCommit msg, int version) + { + return UUIDSerializer.serializer.serializedSize(msg.sessionID, version); + } + }; +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/FinalizePromise.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/FinalizePromise.java b/src/java/org/apache/cassandra/repair/messages/FinalizePromise.java new file mode 100644 index 0000000..6c28347 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/FinalizePromise.java @@ -0,0 +1,95 @@ +/* + * 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.repair.messages; + +import java.io.IOException; +import java.net.InetAddress; +import java.util.UUID; + +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.serializers.InetAddressSerializer; +import org.apache.cassandra.serializers.TypeSerializer; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.UUIDSerializer; + +public class FinalizePromise extends RepairMessage +{ + public final UUID sessionID; + public final InetAddress participant; + public final boolean promised; + + public FinalizePromise(UUID sessionID, InetAddress participant, boolean promised) + { + super(Type.FINALIZE_PROMISE, null); + assert sessionID != null; + assert participant != null; + this.sessionID = sessionID; + this.participant = participant; + this.promised = promised; + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + FinalizePromise that = (FinalizePromise) o; + + if (promised != that.promised) return false; + if (!sessionID.equals(that.sessionID)) return false; + return participant.equals(that.participant); + } + + public int hashCode() + { + int result = sessionID.hashCode(); + result = 31 * result + participant.hashCode(); + result = 31 * result + (promised ? 1 : 0); + return result; + } + + public static MessageSerializer serializer = new MessageSerializer<FinalizePromise>() + { + private TypeSerializer<InetAddress> inetSerializer = InetAddressSerializer.instance; + + public void serialize(FinalizePromise msg, DataOutputPlus out, int version) throws IOException + { + UUIDSerializer.serializer.serialize(msg.sessionID, out, version); + ByteBufferUtil.writeWithShortLength(inetSerializer.serialize(msg.participant), out); + out.writeBoolean(msg.promised); + } + + public FinalizePromise deserialize(DataInputPlus in, int version) throws IOException + { + return new FinalizePromise(UUIDSerializer.serializer.deserialize(in, version), + inetSerializer.deserialize(ByteBufferUtil.readWithShortLength(in)), + in.readBoolean()); + } + + public long serializedSize(FinalizePromise msg, int version) + { + long size = UUIDSerializer.serializer.serializedSize(msg.sessionID, version); + size += ByteBufferUtil.serializedSizeWithShortLength(inetSerializer.serialize(msg.participant)); + size += TypeSizes.sizeof(msg.promised); + return size; + } + }; +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/FinalizePropose.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/FinalizePropose.java b/src/java/org/apache/cassandra/repair/messages/FinalizePropose.java new file mode 100644 index 0000000..c0c49df --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/FinalizePropose.java @@ -0,0 +1,78 @@ +/* + * 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.repair.messages; + +import java.io.IOException; +import java.util.UUID; + +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.utils.UUIDSerializer; + +public class FinalizePropose extends RepairMessage +{ + public final UUID sessionID; + + public FinalizePropose(UUID sessionID) + { + super(Type.FINALIZE_PROPOSE, null); + assert sessionID != null; + this.sessionID = sessionID; + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + FinalizePropose that = (FinalizePropose) o; + + return sessionID.equals(that.sessionID); + } + + public int hashCode() + { + return sessionID.hashCode(); + } + + public String toString() + { + return "FinalizePropose{" + + "sessionID=" + sessionID + + '}'; + } + + public static MessageSerializer serializer = new MessageSerializer<FinalizePropose>() + { + public void serialize(FinalizePropose msg, DataOutputPlus out, int version) throws IOException + { + UUIDSerializer.serializer.serialize(msg.sessionID, out, version); + } + + public FinalizePropose deserialize(DataInputPlus in, int version) throws IOException + { + return new FinalizePropose(UUIDSerializer.serializer.deserialize(in, version)); + } + + public long serializedSize(FinalizePropose msg, int version) + { + return UUIDSerializer.serializer.serializedSize(msg.sessionID, version); + } + }; +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java b/src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java new file mode 100644 index 0000000..57056ef --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java @@ -0,0 +1,124 @@ +/* + * 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.repair.messages; + +import java.io.IOException; +import java.net.InetAddress; +import java.util.HashSet; +import java.util.Set; +import java.util.UUID; + +import com.google.common.collect.ImmutableSet; + +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.serializers.InetAddressSerializer; +import org.apache.cassandra.serializers.TypeSerializer; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.UUIDSerializer; + +public class PrepareConsistentRequest extends RepairMessage +{ + public final UUID parentSession; + public final InetAddress coordinator; + public final Set<InetAddress> participants; + + public PrepareConsistentRequest(UUID parentSession, InetAddress coordinator, Set<InetAddress> participants) + { + super(Type.CONSISTENT_REQUEST, null); + assert parentSession != null; + assert coordinator != null; + assert participants != null && !participants.isEmpty(); + this.parentSession = parentSession; + this.coordinator = coordinator; + this.participants = ImmutableSet.copyOf(participants); + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + PrepareConsistentRequest that = (PrepareConsistentRequest) o; + + if (!parentSession.equals(that.parentSession)) return false; + if (!coordinator.equals(that.coordinator)) return false; + return participants.equals(that.participants); + } + + public int hashCode() + { + int result = parentSession.hashCode(); + result = 31 * result + coordinator.hashCode(); + result = 31 * result + participants.hashCode(); + return result; + } + + public String toString() + { + return "PrepareConsistentRequest{" + + "parentSession=" + parentSession + + ", coordinator=" + coordinator + + ", participants=" + participants + + '}'; + } + + public static MessageSerializer serializer = new MessageSerializer<PrepareConsistentRequest>() + { + private TypeSerializer<InetAddress> inetSerializer = InetAddressSerializer.instance; + + public void serialize(PrepareConsistentRequest request, DataOutputPlus out, int version) throws IOException + { + UUIDSerializer.serializer.serialize(request.parentSession, out, version); + ByteBufferUtil.writeWithShortLength(inetSerializer.serialize(request.coordinator), out); + out.writeInt(request.participants.size()); + for (InetAddress peer : request.participants) + { + ByteBufferUtil.writeWithShortLength(inetSerializer.serialize(peer), out); + } + } + + public PrepareConsistentRequest deserialize(DataInputPlus in, int version) throws IOException + { + UUID sessionId = UUIDSerializer.serializer.deserialize(in, version); + InetAddress coordinator = inetSerializer.deserialize(ByteBufferUtil.readWithShortLength(in)); + int numPeers = in.readInt(); + Set<InetAddress> peers = new HashSet<>(numPeers); + for (int i = 0; i < numPeers; i++) + { + InetAddress peer = inetSerializer.deserialize(ByteBufferUtil.readWithShortLength(in)); + peers.add(peer); + } + return new PrepareConsistentRequest(sessionId, coordinator, peers); + } + + public long serializedSize(PrepareConsistentRequest request, int version) + { + long size = UUIDSerializer.serializer.serializedSize(request.parentSession, version); + size += ByteBufferUtil.serializedSizeWithShortLength(inetSerializer.serialize(request.coordinator)); + size += TypeSizes.sizeof(request.participants.size()); + for (InetAddress peer : request.participants) + { + size += ByteBufferUtil.serializedSizeWithShortLength(inetSerializer.serialize(peer)); + } + return size; + } + }; +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java b/src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java new file mode 100644 index 0000000..cf4410a --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java @@ -0,0 +1,94 @@ +/* + * 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.repair.messages; + +import java.io.IOException; +import java.net.InetAddress; +import java.util.UUID; + +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.serializers.InetAddressSerializer; +import org.apache.cassandra.serializers.TypeSerializer; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.UUIDSerializer; + +public class PrepareConsistentResponse extends RepairMessage +{ + public final UUID parentSession; + public final InetAddress participant; + public final boolean success; + + public PrepareConsistentResponse(UUID parentSession, InetAddress participant, boolean success) + { + super(Type.CONSISTENT_RESPONSE, null); + assert parentSession != null; + assert participant != null; + this.parentSession = parentSession; + this.participant = participant; + this.success = success; + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + PrepareConsistentResponse that = (PrepareConsistentResponse) o; + + if (success != that.success) return false; + if (!parentSession.equals(that.parentSession)) return false; + return participant.equals(that.participant); + } + + public int hashCode() + { + int result = parentSession.hashCode(); + result = 31 * result + participant.hashCode(); + result = 31 * result + (success ? 1 : 0); + return result; + } + + public static MessageSerializer serializer = new MessageSerializer<PrepareConsistentResponse>() + { + private TypeSerializer<InetAddress> inetSerializer = InetAddressSerializer.instance; + public void serialize(PrepareConsistentResponse response, DataOutputPlus out, int version) throws IOException + { + UUIDSerializer.serializer.serialize(response.parentSession, out, version); + ByteBufferUtil.writeWithShortLength(inetSerializer.serialize(response.participant), out); + out.writeBoolean(response.success); + } + + public PrepareConsistentResponse deserialize(DataInputPlus in, int version) throws IOException + { + return new PrepareConsistentResponse(UUIDSerializer.serializer.deserialize(in, version), + inetSerializer.deserialize(ByteBufferUtil.readWithShortLength(in)), + in.readBoolean()); + } + + public long serializedSize(PrepareConsistentResponse response, int version) + { + long size = UUIDSerializer.serializer.serializedSize(response.parentSession, version); + size += ByteBufferUtil.serializedSizeWithShortLength(inetSerializer.serialize(response.participant)); + size += TypeSizes.sizeof(response.success); + return size; + } + }; +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/RepairMessage.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/RepairMessage.java b/src/java/org/apache/cassandra/repair/messages/RepairMessage.java index 55fdb66..3cb913a 100644 --- a/src/java/org/apache/cassandra/repair/messages/RepairMessage.java +++ b/src/java/org/apache/cassandra/repair/messages/RepairMessage.java @@ -37,16 +37,24 @@ public abstract class RepairMessage public static interface MessageSerializer<T extends RepairMessage> extends IVersionedSerializer<T> {} - public static enum Type + public enum Type { VALIDATION_REQUEST(0, ValidationRequest.serializer), VALIDATION_COMPLETE(1, ValidationComplete.serializer), SYNC_REQUEST(2, SyncRequest.serializer), SYNC_COMPLETE(3, SyncComplete.serializer), - ANTICOMPACTION_REQUEST(4, AnticompactionRequest.serializer), PREPARE_MESSAGE(5, PrepareMessage.serializer), SNAPSHOT(6, SnapshotMessage.serializer), - CLEANUP(7, CleanupMessage.serializer); + CLEANUP(7, CleanupMessage.serializer), + + CONSISTENT_REQUEST(8, PrepareConsistentRequest.serializer), + CONSISTENT_RESPONSE(9, PrepareConsistentResponse.serializer), + FINALIZE_PROPOSE(10, FinalizePropose.serializer), + FINALIZE_PROMISE(11, FinalizePromise.serializer), + FINALIZE_COMMIT(12, FinalizeCommit.serializer), + FAILED_SESSION(13, FailSession.serializer), + STATUS_REQUEST(14, StatusRequest.serializer), + STATUS_RESPONSE(15, StatusResponse.serializer); private final byte type; private final MessageSerializer<RepairMessage> serializer; http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/RepairOption.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/RepairOption.java b/src/java/org/apache/cassandra/repair/messages/RepairOption.java index ced6e43..3b13cd8 100644 --- a/src/java/org/apache/cassandra/repair/messages/RepairOption.java +++ b/src/java/org/apache/cassandra/repair/messages/RepairOption.java @@ -237,6 +237,11 @@ public class RepairOption } } + if (option.isIncremental() && !option.isGlobal()) + { + throw new IllegalArgumentException("Incremental repairs cannot be run against a subset of tokens or ranges"); + } + return option; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/StatusRequest.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/StatusRequest.java b/src/java/org/apache/cassandra/repair/messages/StatusRequest.java new file mode 100644 index 0000000..f6a2b82 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/StatusRequest.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.repair.messages; + +import java.io.IOException; +import java.util.UUID; + +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.utils.UUIDSerializer; + +public class StatusRequest extends RepairMessage +{ + public final UUID sessionID; + + public StatusRequest(UUID sessionID) + { + super(Type.STATUS_REQUEST, null); + this.sessionID = sessionID; + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + StatusRequest request = (StatusRequest) o; + + return sessionID.equals(request.sessionID); + } + + public int hashCode() + { + return sessionID.hashCode(); + } + + public String toString() + { + return "StatusRequest{" + + "sessionID=" + sessionID + + '}'; + } + + public static MessageSerializer serializer = new MessageSerializer<StatusRequest>() + { + public void serialize(StatusRequest msg, DataOutputPlus out, int version) throws IOException + { + UUIDSerializer.serializer.serialize(msg.sessionID, out, version); + } + + public StatusRequest deserialize(DataInputPlus in, int version) throws IOException + { + return new StatusRequest(UUIDSerializer.serializer.deserialize(in, version)); + } + + public long serializedSize(StatusRequest msg, int version) + { + return UUIDSerializer.serializer.serializedSize(msg.sessionID, version); + } + }; +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/98d74ed9/src/java/org/apache/cassandra/repair/messages/StatusResponse.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/messages/StatusResponse.java b/src/java/org/apache/cassandra/repair/messages/StatusResponse.java new file mode 100644 index 0000000..99eb76b --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/StatusResponse.java @@ -0,0 +1,90 @@ +/* + * 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.repair.messages; + +import java.io.IOException; +import java.util.UUID; + +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.repair.consistent.ConsistentSession; +import org.apache.cassandra.utils.UUIDSerializer; + +public class StatusResponse extends RepairMessage +{ + public final UUID sessionID; + public final ConsistentSession.State state; + + public StatusResponse(UUID sessionID, ConsistentSession.State state) + { + super(Type.STATUS_RESPONSE, null); + assert sessionID != null; + assert state != null; + this.sessionID = sessionID; + this.state = state; + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + StatusResponse that = (StatusResponse) o; + + if (!sessionID.equals(that.sessionID)) return false; + return state == that.state; + } + + public int hashCode() + { + int result = sessionID.hashCode(); + result = 31 * result + state.hashCode(); + return result; + } + + public String toString() + { + return "StatusResponse{" + + "sessionID=" + sessionID + + ", state=" + state + + '}'; + } + + public static final MessageSerializer serializer = new MessageSerializer<StatusResponse>() + { + public void serialize(StatusResponse msg, DataOutputPlus out, int version) throws IOException + { + UUIDSerializer.serializer.serialize(msg.sessionID, out, version); + out.writeInt(msg.state.ordinal()); + } + + public StatusResponse deserialize(DataInputPlus in, int version) throws IOException + { + return new StatusResponse(UUIDSerializer.serializer.deserialize(in, version), + ConsistentSession.State.valueOf(in.readInt())); + } + + public long serializedSize(StatusResponse msg, int version) + { + return UUIDSerializer.serializer.serializedSize(msg.sessionID, version) + + TypeSizes.sizeof(msg.state.ordinal()); + } + }; +}
