You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2017/02/07 03:11:45 UTC
[4/7] cassandra git commit: Fix consistency of incrementally repaired
data
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());
+ }
+ };
+}