CEP-45: Background reconciliation - replica

patch by Francisco Guerrero; reviewed by Blake Eggleston for CASSANDRA-20378

Author: frankgh

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -6059,6 +6059,11 @@ public static void setDynamicDataMaskingEnabled(boolean enabled)
         }
     }
 
+    public static MutationTrackingSpec getMutationTrackingConfig()
+    {
+        return conf.mutation_tracking;
+    }
+
     public static boolean getMutationTrackingEnabled()
     {
         return conf.mutation_tracking.enabled;

src/java/org/apache/cassandra/config/MutationTrackingSpec.java

@@ -22,4 +22,12 @@ public class MutationTrackingSpec
 {
     public boolean enabled = false;
     public String journal_directory;
+    /**
+     * Whether the background reconciliation process is enabled
+     */
+    public volatile boolean background_reconciliation_enabled = true;
+    /**
+     * The interval in which the backgroun reconciliation process runs
+     */
+    public volatile DurationSpec.LongMillisecondsBound background_reconciliation_interval = new DurationSpec.LongMillisecondsBound("1s");
 }

src/java/org/apache/cassandra/replication/ActiveLogReconciler.java

@@ -173,6 +173,7 @@ public void onFailure(InetAddressAndPort from, RequestFailure failureReason)
             MutationTrackingService.instance().retryFailedWrite(mutationId, toHost, failureReason);
         }
 
+        @Override
         void send()
         {
             RecordPointer pointer = MutationJournal.instance().lookUp(mutationId);

src/java/org/apache/cassandra/replication/CoordinatorLog.java

@@ -271,6 +271,39 @@ Offsets.Immutable collectReplicatedOffsets(boolean persisted)
         }
     }
 
+    /**
+     * @return the computed union of remote-witnessed offsets minus local-witnessed offsets
+     */
+    @Nullable
+    Offsets.Immutable collectLocallyMissingOffsets()
+    {
+        lock.readLock().lock();
+        try
+        {
+            Offsets.Mutable local = witnessedOffsets.get(localNodeId);
+            Offsets.Immutable.Builder missing = null;
+            for (int i = 0; i < participants.size(); i++)
+            {
+                int nodeId = participants.get(i);
+                if (nodeId == localNodeId) continue;
+                Offsets.Immutable diff = Offsets.Immutable.difference(witnessedOffsets.get(nodeId), local);
+                if (!diff.isEmpty())
+                {
+                    if (missing == null)
+                    {
+                        missing = new Offsets.Immutable.Builder(logId);
+                    }
+                    missing.addAll(diff);
+                }
+            }
+            return missing != null ? missing.build() : null;
+        }
+        finally
+        {
+            lock.readLock().unlock();
+        }
+    }
+
     Offsets.Immutable collectReconciledOffsets()
     {
         lock.readLock().lock();

src/java/org/apache/cassandra/replication/MutationSummary.java

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Objects;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
 import org.agrona.collections.Long2ObjectHashMap;
@@ -274,7 +275,8 @@ public int size()
         return summaries.size();
     }
 
-    boolean isEmpty()
+    @VisibleForTesting
+    public boolean isEmpty()
     {
         return size() == 0;
     }

src/java/org/apache/cassandra/replication/MutationTrackingService.java

@@ -49,6 +49,8 @@
 
 import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DurationSpec;
+import org.apache.cassandra.config.MutationTrackingSpec;
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.db.DecoratedKey;
@@ -87,6 +89,7 @@
 import org.apache.cassandra.tcm.ownership.ReplicaGroups;
 import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.MBeanWrapper;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static java.lang.String.format;
@@ -96,19 +99,25 @@
 
 // TODO (expected): persistence (handle restarts)
 // TODO (expected): handle topology changes
-public class MutationTrackingService
+public class MutationTrackingService implements MutationTrackingServiceMBean
 {
+    public static final String MBEAN_NAME = "org.apache.cassandra.db:type=MutationTrackingService";
     public static final String DISABLED_MESSAGE = "Mutation tracking is not enabled. (See mutation_tracking.enabled in cassandra.yaml)";
 
     private static final MutationTrackingService instance;
     private static final ScheduledExecutorPlus executor;
 
+    private static final MutationTrackingSpec config;
+
     static
     {
-        if (DatabaseDescriptor.getMutationTrackingEnabled())
+        config = DatabaseDescriptor.getMutationTrackingConfig();
+
+        if (config.enabled)
         {
             instance = new MutationTrackingService();
             executor = executorFactory().scheduled("Mutation-Tracking-Service", NORMAL);
+            MBeanWrapper.instance.registerMBean(instance, MBEAN_NAME);
         }
         else
         {
@@ -129,12 +138,12 @@ public static MutationTrackingService instance()
 
     public static boolean isEnabled()
     {
-        return DatabaseDescriptor.getMutationTrackingEnabled();
+        return config.enabled;
     }
 
     public static void ensureEnabled()
     {
-        if (!DatabaseDescriptor.getMutationTrackingEnabled())
+        if (!config.enabled)
             throw new IllegalStateException(DISABLED_MESSAGE);
     }
 
@@ -188,6 +197,7 @@ public static void shutdown() throws InterruptedException
     private final ReplicatedOffsetsBroadcaster offsetsBroadcaster = new ReplicatedOffsetsBroadcaster();
     private final LogStatePersister offsetsPersister = new LogStatePersister();
     private final ActiveLogReconciler activeReconciler = new ActiveLogReconciler();
+    private final BackgroundReconciler backgroundReconciler = new BackgroundReconciler();
 
     private final IncomingMutations incomingMutations = new IncomingMutations();
     private final OutgoingMutations outgoingMutations = new OutgoingMutations();
@@ -223,14 +233,53 @@ private synchronized void startInternal(ClusterMetadata metadata)
 
         onNewClusterMetadata(null, metadata);
 
+        if (!keyspaceShards.isEmpty() && !config.background_reconciliation_enabled)
+            logBackgroundReconciliationDisabledWarning(keyspaceShards.keySet());
+
         offsetsBroadcaster.start();
         offsetsPersister.start();
+        backgroundReconciler.start();
 
         ExpiredStatePurger.instance.register(incomingMutations);
 
         started = true;
     }
 
+    @Override
+    public void setMutationTrackingBackgroundReconciliationEnabled(boolean enabled)
+    {
+        if (enabled != config.background_reconciliation_enabled)
+        {
+            logger.info("{} mutation tracking background reconciliation", enabled ? "Enabling" : "Disabling");
+            config.background_reconciliation_enabled = enabled;
+        }
+    }
+
+    @Override
+    public boolean getMutationTrackingBackgroundReconciliationEnabled()
+    {
+        return config.background_reconciliation_enabled;
+    }
+
+    @Override
+    public void setMutationTrackingBackgroundReconciliationIntervalMilliseconds(long intervalMilliseconds)
+    {
+        if (intervalMilliseconds  != config.background_reconciliation_interval.toMilliseconds())
+        {
+            DurationSpec.LongMillisecondsBound backgroundReconciliationInterval =
+            new DurationSpec.LongMillisecondsBound(intervalMilliseconds, TimeUnit.MILLISECONDS);
+            logger.info("Setting mutation tracking background reconciliation interval from {} to {}",
+                        config.background_reconciliation_interval, backgroundReconciliationInterval);
+            config.background_reconciliation_interval = backgroundReconciliationInterval;
+        }
+    }
+
+    @Override
+    public long getMutationTrackingBackgroundReconciliationIntervalMilliseconds()
+    {
+        return config.background_reconciliation_interval.toMilliseconds();
+    }
+
     public void pauseOffsetBroadcast(boolean pause)
     {
         offsetsBroadcaster.pauseOffsetBroadcast(pause);
@@ -862,6 +911,14 @@ private void onNewClusterMetadata(@Nullable ClusterMetadata prev, ClusterMetadat
             // recalculating the shards will repopulate this via the existing callbacks
             log2ShardMap = new ConcurrentHashMap<>();
             keyspaceShards = applyUpdatedMetadata(keyspaceShards, prev, next, this::nextLogId, this::onNewLog);
+
+            if (!config.background_reconciliation_enabled)
+            {
+                Set<String> newKeyspaces = new HashSet<>(keyspaceShards.keySet());
+                newKeyspaces.removeAll(originalKeyspaceShards.keySet());
+                if (!newKeyspaces.isEmpty())
+                    logBackgroundReconciliationDisabledWarning(newKeyspaces);
+            }
         }
         catch (Throwable t)
         {
@@ -1047,6 +1104,12 @@ private static List<SyncTask> unwrapped(Collection<SyncTask> tasks)
         return unwrapped;
     }
 
+    private void logBackgroundReconciliationDisabledWarning(Set<String> keyspaces)
+    {
+        logger.warn("Background reconciliation is disabled but mutation tracking keyspaces exist: {}. " +
+                    "Unreconciled mutations will not be automatically repaired in the background.", keyspaces);
+    }
+
     public static class KeyspaceShards
     {
         private enum UpdateDecision
@@ -1374,6 +1437,93 @@ static int loadHostLogIdFromSystemTable()
         return rows.one().getInt("host_log_id");
     }
 
+    private static class BackgroundReconciler
+    {
+        void start()
+        {
+            scheduleNext();
+        }
+
+        private void scheduleNext()
+        {
+            long intervalMillis = config.background_reconciliation_interval.toMilliseconds();
+            executor.schedule(this::runAndReschedule, intervalMillis, TimeUnit.MILLISECONDS);
+        }
+
+        private void runAndReschedule()
+        {
+            try
+            {
+                run();
+            }
+            finally
+            {
+                scheduleNext();
+            }
+        }
+
+        void run()
+        {
+            MutationTrackingService.instance().forEachKeyspace(this::run);
+        }
+
+        private void run(KeyspaceShards shards)
+        {
+            if (config.background_reconciliation_enabled)
+                shards.forEachShard(this::run);
+        }
+
+        private void run(Shard shard)
+        {
+            try
+            {
+                List<Offsets.Immutable> missing = shard.collectLocallyMissingOffsets();
+                if (missing.isEmpty()) return;
+
+                for (Offsets.Immutable offsets : missing)
+                {
+                    // Prefer pulling from the coordinator
+                    int coordinatorHostId = offsets.logId().hostId();
+                    InetAddressAndPort coordinator = ClusterMetadata.current().directory.endpoint(new NodeId(coordinatorHostId));
+                    InetAddressAndPort pullFrom = FailureDetector.instance.isAlive(coordinator)
+                                                  ? coordinator
+                                                  : findAliveReplica(shard, coordinatorHostId);
+                    if (pullFrom == null)
+                    {
+                        logger.debug("No coordinator or replica is available to process the pull mutation request for missing offset {}",
+                                     offsets);
+                        continue; // No reachable source
+                    }
+
+                    // TODO (expected): backoff, rate limits, per host and total
+                    PullMutationsRequest request = new PullMutationsRequest(offsets);
+                    logger.trace("Requesting pull mutation request from replica {} for missing offset {}", pullFrom, offsets);
+                    MessagingService.instance().send(Message.out(Verb.PULL_MUTATIONS_REQ, request), pullFrom);
+                }
+            }
+            catch (Throwable throwable)
+            {
+                // Avoid throwing an exception in the reconciliation step to prevent the scheduled task from
+                // being killed
+                logger.error("Exception encountered during background reconciliation of shard={}", shard, throwable);
+            }
+        }
+
+        private InetAddressAndPort findAliveReplica(Shard shard, int excludeHostId)
+        {
+            for (InetAddressAndPort replica : shard.remoteReplicas())
+            {
+                int replicaId = ClusterMetadata.current().directory.peerId(replica).id();
+                if (replicaId != excludeHostId && FailureDetector.instance.isAlive(replica))
+                {
+                    logger.trace("Found alive replica {} with replica id {}", replica, replicaId);
+                    return replica;
+                }
+            }
+            return null;
+        }
+    }
+
     // TODO (later): a more intelligent heuristic for offsets included in broadcasts
     private static class ReplicatedOffsetsBroadcaster
     {
@@ -1509,6 +1659,24 @@ public void resumeActiveReconcilerRegularPriority()
         activeReconciler.resumeRegularPriorityForTesting();
     }
 
+    @VisibleForTesting
+    public void reconcileForTesting()
+    {
+        backgroundReconciler.run();
+    }
+
+    @VisibleForTesting
+    public void pauseBackgroundReconciler()
+    {
+        config.background_reconciliation_enabled = false;
+    }
+
+    @VisibleForTesting
+    public void resumeBackgroundReconciler()
+    {
+        config.background_reconciliation_enabled = true;
+    }
+
     @VisibleForTesting
     public static class TestAccess
     {