src/java/org/apache/cassandra/hints/HintsDispatcher.java

/*
 * 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.hints;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BooleanSupplier;
import java.util.function.Function;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.cassandra.concurrent.DebuggableTask.RunnableDebuggableTask;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.exceptions.RetryOnDifferentSystemException;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.metrics.HintsServiceMetrics;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.service.PreserveTimestamp;
import org.apache.cassandra.service.accord.IAccordService.IAccordResult;
import org.apache.cassandra.service.accord.txn.TxnResult;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationMutationHelper;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationMutationHelper.SplitMutation;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.MonotonicClock;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.concurrent.Condition;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static org.apache.cassandra.hints.HintsDispatcher.Callback.Outcome.FAILURE;
import static org.apache.cassandra.hints.HintsDispatcher.Callback.Outcome.INTERRUPTED;
import static org.apache.cassandra.hints.HintsDispatcher.Callback.Outcome.RETRY_DIFFERENT_SYSTEM;
import static org.apache.cassandra.hints.HintsDispatcher.Callback.Outcome.SUCCESS;
import static org.apache.cassandra.hints.HintsDispatcher.Callback.Outcome.TIMEOUT;
import static org.apache.cassandra.hints.HintsService.RETRY_ON_DIFFERENT_SYSTEM_UUID;
import static org.apache.cassandra.metrics.HintsServiceMetrics.ACCORD_HINT_ENDPOINT;
import static org.apache.cassandra.metrics.HintsServiceMetrics.updateDelayMetrics;
import static org.apache.cassandra.net.Verb.HINT_REQ;
import static org.apache.cassandra.service.accord.txn.TxnResult.Kind.retry_new_protocol;
import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime;
import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition;

/**
 * Dispatches a single hints file to a specified node in a batched manner.
 *
 * Uses either {@link HintMessage.Encoded} - when dispatching hints into a node with the same messaging version as the hints file,
 * or {@link HintMessage}, when conversion is required.
 */
final class HintsDispatcher implements AutoCloseable
{
    private static final Logger logger = LoggerFactory.getLogger(HintsDispatcher.class);
    private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES);

    private enum Action { CONTINUE, ABORT }

    private final HintsReader reader;
    final UUID hostId;

    @Nullable
    final InetAddressAndPort address;
    private final int messagingVersion;
    private final BooleanSupplier abortRequested;

    private InputPosition currentPagePosition;

    // Hints from the batch log that were attempted on Accord don't have a list of hosts that need hinting
    // since Accord doesn't expose that on failure. If Accord no longer manages the range for this hint then we need
    // to send the hint to all replicas after the page succeeds
    private final Queue<Hint> hintsNeedingRehinting = new LinkedList<>();

    private HintsDispatcher(HintsReader reader, UUID hostId, @Nullable InetAddressAndPort address, int messagingVersion, BooleanSupplier abortRequested)
    {
        checkArgument(address != null ^ hostId.equals(RETRY_ON_DIFFERENT_SYSTEM_UUID), "address must be nonnull or hostId must be " + RETRY_ON_DIFFERENT_SYSTEM_UUID);
        currentPagePosition = null;

        this.reader = reader;
        this.hostId = hostId;
        this.address = address;
        this.messagingVersion = messagingVersion;
        this.abortRequested = abortRequested;
    }

    static HintsDispatcher create(File file, RateLimiter rateLimiter, @Nullable InetAddressAndPort address, UUID hostId, BooleanSupplier abortRequested)
    {
        int messagingVersion = address == null ? MessagingService.current_version : MessagingService.instance().versions.get(address);
        HintsDispatcher dispatcher = new HintsDispatcher(HintsReader.open(file, rateLimiter), hostId, address, messagingVersion, abortRequested);
        HintDiagnostics.dispatcherCreated(dispatcher);
        return dispatcher;
    }

    public void close()
    {
        HintDiagnostics.dispatcherClosed(this);
        reader.close();
    }

    String destination()
    {
        return address == null ? "RETRY_ON_DIFFERENT_SYSTEM" : address.toString();
    }

    void seek(InputPosition position)
    {
        reader.seek(position);
    }

    /**
     * @return whether or not dispatch completed entirely and successfully
     */
    boolean dispatch()
    {
        for (HintsReader.Page page : reader)
        {
            currentPagePosition = page.position;
            if (dispatch(page) != Action.CONTINUE)
                return false;
        }

        return true;
    }

    /**
     * @return offset of the first non-delivered page
     */
    InputPosition dispatchPosition()
    {
        return currentPagePosition;
    }


    // retry in case of a timeout; stop in case of a failure, host going down, or delivery paused
    private Action dispatch(HintsReader.Page page)
    {
        HintDiagnostics.dispatchPage(this);
        return sendHintsAndAwait(page);
    }

    private Action sendHintsAndAwait(HintsReader.Page page)
    {
        try
        {
            return doSendHintsAndAwait(page, null);
        }
        finally
        {
            hintsNeedingRehinting.clear();
        }
    }

    private Action doSendHintsAndAwait(HintsReader.Page page, @Nullable BitSet hintsFilter)
    {
        List<Callback> callbacks = new ArrayList<>();

        /*
         * If hints file messaging version matches the version of the target host, we'll use the optimised path -
         * skipping the redundant decoding/encoding cycle of the already encoded hint.
         *
         * If that is not the case, we'll need to perform conversion to a newer (or an older) format, and decoding the hint
         * is an unavoidable intermediate step.
         *
         * If these hints are from the batchlog and were originally attempted on Accord then
         * we also need to decode so we can route the Hint contents appropriately.
         *
         * If filtering of hints is requested, because this is retrying a page that had some retry on different system
         * errors, then also don't go down the sendEncodedHints path since it won't re-route the mutation and will trigger
         * the same retry on different system error.
         */
        boolean isBatchLogHints = hostId.equals(RETRY_ON_DIFFERENT_SYSTEM_UUID);
        boolean sendEncodedHints = reader.descriptor().messagingVersion() == messagingVersion && !isBatchLogHints && hintsFilter == null;
        // If the hints filter is set then splitting the hints is needed and encoded hints can't do that
        checkState(!sendEncodedHints || hintsFilter == null, "Should not send encoded hints if hints filter is set");
        Action action = sendEncodedHints
                      ? sendHints(page.buffersIterator(), null, callbacks, this::sendEncodedHint)
                      : sendHints(page.hintsIterator(), hintsFilter, callbacks, this::sendHint);

        if (action == Action.ABORT)
            return action;

        BitSet retryDifferentSystemHints = new BitSet(callbacks.size());
        long success = 0, failures = 0, timeouts = 0, retryDifferentSystem = 0;
        for (int i = 0; i < callbacks.size(); i++)
        {
            Callback cb = callbacks.get(i);
            Callback.Outcome outcome = cb.await();
            if (outcome == Callback.Outcome.SUCCESS) success++;
            else if (outcome == Callback.Outcome.FAILURE) failures++;
            else if (outcome == Callback.Outcome.TIMEOUT) timeouts++;
            else if (outcome == RETRY_DIFFERENT_SYSTEM)
            {
                retryDifferentSystemHints.set(i);
                retryDifferentSystem++;
            }
            else throw new IllegalStateException("Unhandled outcome: " + outcome);
        }

        updateMetrics(success, failures, timeouts, retryDifferentSystem);

        // If the only errors were retryDifferentSystem and we aren't already filtering the hints then retry
        // immediately otherwise we will repeat the page later including any successful hints we may have already delivered
        // Hints for the batch log can hit RETRY_DIFFERENT_SYSTEM but don't need to be retried here and it could result
        // in the same hint ending up in hintsNeedingRehinting twice
        boolean failedRetryDifferentSystem = false;
        if (retryDifferentSystem > 0 && failures < 1 && timeouts < 1 && hintsFilter == null && !isBatchLogHints)
        {
            reader.seek(currentPagePosition);
            Action retryResult = doSendHintsAndAwait(page, retryDifferentSystemHints);
            if (retryResult != Action.CONTINUE)
                failedRetryDifferentSystem = true;
        }

        // The batchlog Accord hints need to return abort if any hint needs to be retried and retry the whole page
        // since we don't want hints to ping pong back and forth via hintsNeedingRehinting
        if (failures > 0 || timeouts > 0 || failedRetryDifferentSystem || (isBatchLogHints && retryDifferentSystem > 0))
        {
            HintDiagnostics.pageFailureResult(this, success, failures, timeouts, retryDifferentSystem);
            return Action.ABORT;
        }
        else
        {
            HintDiagnostics.pageSuccessResult(this, success, failures, timeouts, retryDifferentSystem);
            rehintHintsNeedingRehinting();
            return Action.CONTINUE;
        }
    }

    private void rehintHintsNeedingRehinting()
    {
        ClusterMetadata cm = ClusterMetadata.current();
        Hint hint;
        while ((hint = hintsNeedingRehinting.poll()) != null)
        {
            HintsService.instance.writeForAllReplicas(hint);
            Mutation mutation = hint.mutation;
            // Also may need to apply locally because it's possible this is from the batchlog
            // and we never applied it locally
            // TODO (review): Additional error handling necessary? Hints are lossy
            DataPlacement dataPlacement = cm.placements.get(cm.schema.getKeyspace(mutation.getKeyspaceName()).getMetadata().params.replication);
            VersionedEndpoints.ForToken forToken = dataPlacement.writes.forToken(mutation.key().getToken());
            Replica self = forToken.get().selfIfPresent();
            if (self != null)
            {
                Stage.MUTATION.maybeExecuteImmediately(new RunnableDebuggableTask()
                {
                    private final long approxCreationTimeNanos = MonotonicClock.Global.approxTime.now();
                    private volatile long approxStartTimeNanos;

                    @Override
                    public void run()
                    {
                        approxStartTimeNanos = MonotonicClock.Global.approxTime.now();
                        mutation.apply();
                    }

                    @Override
                    public long creationTimeNanos()
                    {
                        return approxCreationTimeNanos;
                    }

                    @Override
                    public long startTimeNanos()
                    {
                        return approxStartTimeNanos;
                    }

                    @Override
                    public String description()
                    {
                        return "HintsService rehinting Accord txn";
                    }
                });
            }
        }

    }

    private void updateMetrics(long success, long failures, long timeouts, long retryDifferentSystem)
    {
        HintsServiceMetrics.hintsSucceeded.mark(success);
        HintsServiceMetrics.hintsFailed.mark(failures);
        HintsServiceMetrics.hintsTimedOut.mark(timeouts);
        HintsServiceMetrics.hintsRetryDifferentSystem.mark(retryDifferentSystem);
    }

    /*
     * Sending hints in compatibility mode.
     */
    private <T> Action sendHints(Iterator<T> hints, @Nullable BitSet hintsFilter, Collection<Callback> callbacks, Function<T, Callback> sendFunction)
    {
        int hintIndex = -1;
        while (hints.hasNext())
        {
            if (abortRequested.getAsBoolean())
            {
                HintDiagnostics.abortRequested(this);
                return Action.ABORT;
            }

            T hint = hints.next();
            hintIndex++;
            if (hintsFilter != null && !hintsFilter.get(hintIndex))
                continue;

            callbacks.add(sendFunction.apply(hint));
        }
        return Action.CONTINUE;
    }

    private Callback sendHint(Hint hint)
    {
        ClusterMetadata cm = ClusterMetadata.current();
        SplitHint splitHint = splitHintIntoAccordAndNormal(cm, hint);
        Mutation accordHintMutation = splitHint.accordMutation;
        Dispatcher.RequestTime requestTime;
        IAccordResult<TxnResult> accordTxnResult = null;
        if (accordHintMutation != null)
        {
            requestTime = Dispatcher.RequestTime.forImmediateExecution();
            accordTxnResult = accordHintMutation != null ? ConsensusMigrationMutationHelper.instance().mutateWithAccordAsync(cm, accordHintMutation, null, requestTime, PreserveTimestamp.yes) : null;
        }

        Hint normalHint = splitHint.normalHint;
        Callback callback = new Callback(address, hint.creationTime, accordTxnResult);
        if (normalHint != null)
        {
            // We had a hint that was supposed to be done on Accord for the batch log (otherwise address would be non-null),
            // but Accord no longer manages that table/range and now we don't know which nodes (if any) are missing the Mutation.
            // Convert them to per replica hints *after* all the hints in this page have been applied so we can be reasonably sure
            // this page isn't going to be played again thus avoiding any futher amplification from the same hint being
            // replayed and repeatedly converted to per replica hints
            if (address == null)
            {
                checkState(hostId.equals(RETRY_ON_DIFFERENT_SYSTEM_UUID), "If there is no address to send the hint to then the host ID should be BATCHLOG_ACCORD_HINT_UUID");
                callback.onResponse(null);
                hintsNeedingRehinting.add(normalHint);
            }
            else
            {
                Message<?> message = Message.out(HINT_REQ, new HintMessage(hostId, normalHint));
                MessagingService.instance().sendWithCallback(message, address, callback);
            }
        }
        else
        {
            // Don't wait for a normal response that will never come since no hints were sent
            callback.onResponse(null);
        }

        return callback;
    }

    /**
     * Result of splitting a hint across Accord and non-transactional boundaries
     */
    private class SplitHint
    {
        private final Mutation accordMutation;
        private final Hint normalHint;

        public SplitHint(Mutation accordMutation, Hint normalHint)
        {
            this.accordMutation = accordMutation;
            this.normalHint = normalHint;
        }

        @Override
        public String toString()
        {
            return "SplitHint{" +
                   "accordMutation=" + accordMutation +
                   ", normalHint=" + normalHint +
                   '}';
        }
    }

    private SplitHint splitHintIntoAccordAndNormal(ClusterMetadata cm, Hint hint)
    {
        SplitMutation<Mutation> splitMutation = ConsensusMigrationMutationHelper.instance().splitMutationIntoAccordAndNormal(hint.mutation, cm);
        if (splitMutation.accordMutation == null)
            return new SplitHint(null, hint);
        if (splitMutation.normalMutation == null)
            return new SplitHint(splitMutation.accordMutation, null);
        Hint normalHint = Hint.create(splitMutation.normalMutation, hint.creationTime, splitMutation.normalMutation.smallestGCGS());
        return new SplitHint(splitMutation.accordMutation, normalHint);
    }

    /*
     * Sending hints in raw mode.
     */

    private Callback sendEncodedHint(ByteBuffer hint)
    {
        HintMessage.Encoded message = new HintMessage.Encoded(hostId, hint, messagingVersion);
        Callback callback = new Callback(address, message.getHintCreationTime());
        MessagingService.instance().sendWithCallback(Message.out(HINT_REQ, message), address, callback);
        return callback;
    }

    static final class Callback implements RequestCallback, BiConsumer<TxnResult, Throwable>
    {
        enum Outcome { SUCCESS, TIMEOUT, FAILURE, INTERRUPTED, RETRY_DIFFERENT_SYSTEM }

        private final long start = approxTime.now();
        private final Condition condition = newOneTimeCondition();
        private Outcome normalOutcome;
        private Outcome accordOutcome;
        @Nullable
        private final InetAddressAndPort to;
        private final long hintCreationNanoTime;

        private Callback(@Nonnull InetAddressAndPort to, long hintCreationTimeMillisSinceEpoch)
        {
            this(to, hintCreationTimeMillisSinceEpoch, null);
        }

        private Callback(@Nullable InetAddressAndPort to, long hintCreationTimeMillisSinceEpoch, @Nullable IAccordResult<TxnResult> accordTxnResult)
        {
            this.to = to != null ? to : ACCORD_HINT_ENDPOINT;
            this.hintCreationNanoTime = approxTime.translate().fromMillisSinceEpoch(hintCreationTimeMillisSinceEpoch);
            if (accordTxnResult != null)
                accordTxnResult.addCallback(this);
            else
                accordOutcome = SUCCESS;
        }

        Outcome await()
        {
            boolean timedOut;
            try
            {
                timedOut = !condition.awaitUntil(HINT_REQ.expiresAtNanos(start));
            }
            catch (InterruptedException e)
            {
                logger.warn("Hint dispatch was interrupted", e);
                return INTERRUPTED;
            }
            normalOutcome = timedOut ? TIMEOUT : normalOutcome;

            return outcome();
        }

        private Outcome outcome()
        {
            checkState((normalOutcome != null && accordOutcome != null) || (normalOutcome != SUCCESS || accordOutcome != SUCCESS), "Outcome for both normal and accord hint delivery should be known");
            if (normalOutcome == RETRY_DIFFERENT_SYSTEM || accordOutcome == RETRY_DIFFERENT_SYSTEM)
                return RETRY_DIFFERENT_SYSTEM;
            if (normalOutcome == TIMEOUT || accordOutcome == TIMEOUT)
                return TIMEOUT;
            if (normalOutcome == FAILURE || accordOutcome == FAILURE)
                return FAILURE;
            checkState(normalOutcome == SUCCESS && accordOutcome == SUCCESS, "Hint delivery should have been successful");
            return SUCCESS;
        }

        private synchronized void maybeSignal()
        {
            if ((normalOutcome != null && accordOutcome != null) || normalOutcome == FAILURE || accordOutcome == FAILURE)
            {
                updateDelayMetrics(to, approxTime.now() - this.hintCreationNanoTime);
                condition.signalAll();
            }
        }

        @Override
        public boolean invokeOnFailure()
        {
            return true;
        }

        @Override
        public void onFailure(InetAddressAndPort from, RequestFailure failureMessage)
        {
            if (failureMessage.reason == RequestFailureReason.RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM)
                normalOutcome = RETRY_DIFFERENT_SYSTEM;
            else
                normalOutcome = FAILURE;
            maybeSignal();
        }

        @Override
        public void onResponse(Message msg)
        {
            normalOutcome = SUCCESS;
            maybeSignal();
        }

        @Override
        public void accept(TxnResult success, Throwable fail)
        {
            if (fail != null)
            {
                if (fail instanceof RequestExecutionException || fail instanceof RetryOnDifferentSystemException)
                {
                    if (fail instanceof RetryOnDifferentSystemException)
                        accordOutcome = RETRY_DIFFERENT_SYSTEM;
                    else
                        accordOutcome = TIMEOUT;
                    String msg = "Accord hint delivery transaction failed retriably";
                    if (noSpamLogger.getStatement(msg).shouldLog(Clock.Global.nanoTime()))
                        logger.error(msg, fail);
                }
                else
                {
                    accordOutcome = FAILURE;
                    String msg = "Accord hint delivery transaction failed permanently";
                    if (noSpamLogger.getStatement(msg).shouldLog(Clock.Global.nanoTime()))
                        logger.error(msg, fail);
                }
            }
            else
            {
                if (success.kind() == retry_new_protocol)
                    accordOutcome = RETRY_DIFFERENT_SYSTEM;
                else
                    accordOutcome = SUCCESS;
            }
            maybeSignal();
        }
    }
}