diff --git a/CHANGES.txt b/CHANGES.txt index 0942f1212b..8ae621f938 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Always send TCM commit failures as Messaging failures (CASSANDRA-21457) * Fix ReadCommand serializedSize() using incorrect epoch (CASSANDRA-21438) * Allocation improvements in ProtocolVersion, StorageProxy and MerkleTree (CASSANDRA-21199) * Don’t leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236) diff --git a/src/java/org/apache/cassandra/net/Message.java b/src/java/org/apache/cassandra/net/Message.java index 41f0281706..14b510937f 100644 --- a/src/java/org/apache/cassandra/net/Message.java +++ b/src/java/org/apache/cassandra/net/Message.java @@ -254,6 +254,12 @@ public class Message implements ResponseContext return outWithParam(nextId(), verb, 0, payload, flag.addTo(0), null, null); } + public static Message outWithFlag(Verb verb, T payload, MessageFlag flag, long expireAtNanos) + { + assert !verb.isResponse(); + return outWithParam(nextId(), verb, expireAtNanos, payload, flag.addTo(0), null, null); + } + public static Message outWithFlag(Verb verb, T payload, Dispatcher.RequestTime requestTime, MessageFlag flag) { return outWithFlags(verb, payload, requestTime, flag.addTo(0)); diff --git a/src/java/org/apache/cassandra/tcm/Commit.java b/src/java/org/apache/cassandra/tcm/Commit.java index f002b3e66e..6baf00b6aa 100644 --- a/src/java/org/apache/cassandra/tcm/Commit.java +++ b/src/java/org/apache/cassandra/tcm/Commit.java @@ -27,6 +27,8 @@ import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import accord.utils.Invariants; + import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.exceptions.ExceptionCode; import org.apache.cassandra.io.IVersionedSerializer; @@ -281,7 +283,7 @@ public class Commit } else { - assert t instanceof Failure; + Invariants.require(t instanceof Failure); Failure failure = (Failure) t; out.writeByte(failure.rejected ? REJECTED : FAILED); out.writeUnsignedVInt32(failure.code.value); @@ -327,9 +329,10 @@ public class Commit } else { - assert t instanceof Failure; - size += VIntCoding.computeUnsignedVIntSize(((Failure) t).code.value); - size += TypeSizes.sizeof(((Failure)t).message); + Invariants.require(t instanceof Failure); + Failure failure = (Failure) t; + size += VIntCoding.computeUnsignedVIntSize(failure.code.value); + size += TypeSizes.sizeof(failure.message); size += VIntCoding.computeUnsignedVIntSize(serializationVersion.asInt()); size += LogState.metadataSerializer.serializedSize(t.logState(), serializationVersion); } @@ -375,15 +378,9 @@ public class Commit Result.Success success = result.success(); replicator.send(success, message.from()); logger.info("Responding with full result {} to sender {}", result, message.from()); - // TODO: this response message can get lost; how do we re-discover this on the other side? - // TODO: what if we have holes after replaying? - messagingService.accept(message.responseWith(result), message.from()); - } - else - { - Result.Failure failure = result.failure(); - messagingService.accept(message.responseWith(failure), message.from()); } + + messagingService.accept(message.responseWith(result), message.from()); } private void checkCMSState() diff --git a/src/java/org/apache/cassandra/tcm/PeerLogFetcher.java b/src/java/org/apache/cassandra/tcm/PeerLogFetcher.java index ceac4c7025..12cdc18e5d 100644 --- a/src/java/org/apache/cassandra/tcm/PeerLogFetcher.java +++ b/src/java/org/apache/cassandra/tcm/PeerLogFetcher.java @@ -92,11 +92,11 @@ public class PeerLogFetcher logger.info("Fetching log from {}, at least {}", remote, awaitAtleast); try (Timer.Context ctx = TCMMetrics.instance.fetchPeerLogLatency.time()) { - RemoteProcessor.sendWithCallbackAsync(fetchRes, - Verb.TCM_FETCH_PEER_LOG_REQ, - new FetchPeerLog(before), - new RemoteProcessor.CandidateIterator(Collections.singletonList(remote), false), - Retry.untilElapsed(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS), TCMMetrics.instance.fetchLogRetries)); + RemoteProcessor.sendWithRetries(Verb.TCM_FETCH_PEER_LOG_REQ, + new FetchPeerLog(before), + fetchRes, + new RemoteProcessor.CandidateIterator(Collections.singletonList(remote), false), + Retry.untilElapsed(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS), TCMMetrics.instance.fetchLogRetries)); return fetchRes.map((logState) -> { log.append(logState); diff --git a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java index 59b63d42e1..b14e73feb8 100644 --- a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java +++ b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java @@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.Consumer; import java.util.function.Supplier; import com.codahale.metrics.Timer; @@ -34,6 +35,7 @@ import com.codahale.metrics.Timer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.exceptions.RequestFailure; import org.apache.cassandra.exceptions.RequestFailureReason; @@ -42,15 +44,16 @@ import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.metrics.TCMMetrics; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessageDelivery; +import org.apache.cassandra.net.MessageFlag; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.RequestCallbackWithFailure; import org.apache.cassandra.net.Verb; -import org.apache.cassandra.service.WaitStrategy; import org.apache.cassandra.tcm.Discovery.DiscoveredNodes; import org.apache.cassandra.tcm.log.Entry; import org.apache.cassandra.tcm.log.LocalLog; import org.apache.cassandra.tcm.log.LogState; import org.apache.cassandra.utils.AbstractIterator; +import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.concurrent.AsyncPromise; import org.apache.cassandra.utils.concurrent.Future; @@ -78,10 +81,14 @@ public final class RemoteProcessor implements Processor { try { - Commit.Result result = sendWithCallback(Verb.TCM_COMMIT_REQ, - new Commit(entryId, transform, lastKnown), - new CandidateIterator(candidates(false)), - retryPolicy); + Commit.Result result = sendWithRetries(Verb.TCM_COMMIT_REQ, + new Commit(entryId, transform, lastKnown), + (Commit.Result res) -> { + if (res.isFailure() && !res.failure().rejected) + throw new IllegalArgumentException(res.failure().message); + }, + new CandidateIterator(candidates(false)), + retryPolicy); log.append(result.logState()); @@ -171,11 +178,11 @@ public final class RemoteProcessor implements Processor { Promise remoteRequest = new AsyncPromise<>(); Epoch currentEpoch = log.metadata().epoch; - sendWithCallbackAsync(remoteRequest, - Verb.TCM_FETCH_CMS_LOG_REQ, - new FetchCMSLog(currentEpoch, ClusterMetadataService.state() == REMOTE), - candidates, - Retry.withNoTimeLimit(TCMMetrics.instance.fetchLogRetries)); + sendWithRetries(Verb.TCM_FETCH_CMS_LOG_REQ, + new FetchCMSLog(currentEpoch, ClusterMetadataService.state() == REMOTE), + remoteRequest, + candidates, + Retry.withNoTimeLimit(TCMMetrics.instance.fetchLogRetries)); return remoteRequest.map((replay) -> { if (!replay.isEmpty()) { @@ -188,14 +195,13 @@ public final class RemoteProcessor implements Processor } } - // todo rename to send with retries or something - public static RSP sendWithCallback(Verb verb, REQ request, CandidateIterator candidates, WaitStrategy backoff) + public static RSP sendWithRetries(Verb verb, REQ request, Consumer check, CandidateIterator candidates, Retry retry) { + Promise future = AsyncPromise.uncancellable(); + sendWithRetries(verb, request, check, future, candidates, retry); try { - Promise promise = new AsyncPromise<>(); - sendWithCallbackAsync(promise, verb, request, candidates, backoff); - return promise.await().get(); + return future.get(); } catch (InterruptedException | ExecutionException e) { @@ -203,47 +209,92 @@ public final class RemoteProcessor implements Processor } } - public static void sendWithCallbackAsync(Promise promise, Verb verb, REQ request, CandidateIterator candidates, WaitStrategy backoff) + public static void sendWithRetries(Verb verb, REQ request, Promise future, CandidateIterator candidates, Retry retry) { - //TODO (now): the retry defines how long to wait for a retry, but the old behavior scheduled the message right away... should this be delayed as well? - MessagingService.instance().sendWithRetries(backoff, MessageDelivery.ImmediateRetryScheduler.instance, - verb, request, candidates, - (attempt, success, failure) -> { - if (failure != null) promise.tryFailure(failure); - else promise.trySuccess(success.payload); - }, - (attempt, from, failure) -> { - if (promise.isDone() || promise.isCancelled()) - return false; - if (failure.reason == RequestFailureReason.NOT_CMS) - { - logger.debug("{} is not a member of the CMS, querying it to discover current membership", from); - DiscoveredNodes cms = tryDiscover(from); - candidates.addCandidates(cms); - candidates.timeout(from); - logger.debug("Got CMS from {}: {}, retrying on: {}", from, cms, candidates); - } - else - { - candidates.timeout(from); - logger.warn("Got error from {}: {} when sending {}, retrying on {}", from, failure, verb, candidates); - } - return true; - }, - (attempt, reason, from, failure) -> { - switch (reason) - { - case NoMoreCandidates: - return String.format("Ran out of candidates while sending %s: %s", verb, candidates); - case GiveUp: - return String.format("Could not succeed sending %s to %s; policy %s gave up", verb, candidates, backoff); - case Interrupted: - case FailedSchedule: - return null; - default: - throw new UnsupportedOperationException(reason.name()); - } - }); + sendWithRetries(verb, request, ignore_ -> {}, future, candidates, retry); + } + + /** + * Sends a request to given candidate nodes with retries, respecting the retry policy. + * + * If request handler expects node to be CMS, handles CMS discovery if a candidate node reports it's not a CMS member. + */ + public static void sendWithRetries(Verb verb, REQ request, Consumer check, Promise future, CandidateIterator candidates, Retry retry) + { + if (!candidates.hasNext()) + { + future.setFailure(new MessageDelivery.NoMoreCandidatesException(String.format("Ran out of candidates while sending %s: %s", verb, candidates))); + return; + } + else if (retry.hasExpired()) + { + future.setFailure(new MessageDelivery.GivingUpException(retry.attempts(), String.format("Could not succeed sending %s to %s; policy %s gave up", verb, candidates, retry))); + return; + } + + InetAddressAndPort candidate = candidates.next(); + long waitNanos = Math.min(verb.expiresAfterNanos(), Math.max(0, retry.remainingNanos())); + Message msg = Message.outWithFlag(verb, request, MessageFlag.CALL_BACK_ON_FAILURE, Clock.Global.nanoTime() + waitNanos); + MessagingService.instance().sendWithCallback(msg, candidate, new RequestCallbackWithFailure() + { + @Override + public void onFailure(InetAddressAndPort from, RequestFailure failure) + { + switch (failure.reason) + { + case UNKNOWN: + candidates.timeout(candidate); + logger.warn("Got error from {}: {} when sending {}, retrying on {}", from, failure, verb, candidates); + break; + case TIMEOUT: + candidates.timeout(candidate); + logger.warn("Got error from {}: timeout when sending {}, retrying on {}", candidate, verb, candidates); + break; + case NOT_CMS: + logger.debug("{} is not a member of the CMS, querying it to discover current membership", from); + DiscoveredNodes cms = tryDiscover(from); + candidates.addCandidates(cms); + candidates.timeout(from); + logger.debug("Got CMS from {}: {}, retrying on: {}", from, cms, candidates); + break; + default: + // Unknown exception - add candidate back and retry + candidates.timeout(candidate); + logger.warn("Unexpected error ({}) sending {} to {}, retrying", failure, verb, candidate); + } + + if (Thread.currentThread().isInterrupted()) // preserve interrupt status + { + Thread.currentThread().interrupt(); + future.setFailure(new InterruptedException("Interrupted while sending " + verb)); + } + else + { + long waitFor = retry.computeWait(); + if (waitFor < 0) + { + // The retry strategy returns a negative wait to signal that its attempt budget is exhausted. + future.setFailure(new MessageDelivery.GivingUpException(retry.attempts(), String.format("Could not succeed sending %s to %s; policy %s exhausted attempts", verb, candidates, retry))); + return; + } + ScheduledExecutors.nonPeriodicTasks.schedule(() -> sendWithRetries(verb, request, check, future, candidates, retry), waitFor, TimeUnit.MILLISECONDS); + } + } + + @Override + public void onResponse(Message msg) + { + try + { + check.accept(msg.payload); + future.setSuccess(msg.payload); + } + catch (Throwable t) + { + onFailure(msg.from(), new RequestFailure(RequestFailureReason.UNKNOWN, t)); + } + } + }); } private static DiscoveredNodes tryDiscover(InetAddressAndPort ep) diff --git a/src/java/org/apache/cassandra/tcm/Retry.java b/src/java/org/apache/cassandra/tcm/Retry.java index f4951bf110..be79f10f26 100644 --- a/src/java/org/apache/cassandra/tcm/Retry.java +++ b/src/java/org/apache/cassandra/tcm/Retry.java @@ -85,6 +85,11 @@ public class Retry implements WaitStrategy return true; } + public long computeWait() + { + return computeWait(attempts, TimeUnit.MILLISECONDS); + } + @Override public long computeWaitUntil(int attempts) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/CommitStartupByNonCMSNodeTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/CommitStartupByNonCMSNodeTest.java new file mode 100644 index 0000000000..cda51ac274 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/log/CommitStartupByNonCMSNodeTest.java @@ -0,0 +1,83 @@ +/* + * 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.distributed.test.log; + +import java.util.Random; +import java.util.function.Supplier; + +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.shared.ClusterUtils; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.membership.NodeAddresses; +import org.apache.cassandra.tcm.membership.NodeVersion; +import org.apache.cassandra.tcm.transformations.Startup; +import org.apache.cassandra.utils.FBUtilities; + +import static org.junit.Assert.fail; + +public class CommitStartupByNonCMSNodeTest extends FuzzTestBase +{ + @Test + public void commitStartupByNonCMSNode() throws Throwable + { + try (Cluster cluster = Cluster.build(4) + .withConfig(conf -> conf.set("request_timeout", "1000ms") // for TCM commit + .set("cms_retry_delay", "10ms,retries=1") // avoid retries by CMS node to avoid triggering log fetch + .set("write_request_timeout", "100ms") // time out paxos writes quickly + .with(Feature.NETWORK, Feature.GOSSIP)) + .start()) + { + cluster.setUncaughtExceptionsFilter(t -> t.getMessage() != null && t.getMessage().contains("There are not enough nodes in dc0 datacenter to satisfy replication factor")); + Random rnd = new Random(2); + Supplier nodeSelector = () -> rnd.nextInt(cluster.size() - 1) + 1; + cluster.get(nodeSelector.get()).nodetoolResult("cms", "reconfigure", "3").asserts().success(); + for (int i = 2; i <= 3; i++) + ClusterUtils.stopUnchecked(cluster.get(i)); + + Thread startNode2 = new Thread(() -> { + cluster.get(2).startup(); + }); + + Thread commitStartupTranformation = new Thread(() -> { + cluster.get(4).runOnInstance(() -> { + try + { + ClusterMetadataService.instance().commit(new Startup(ClusterMetadata.current().myNodeId(), + new NodeAddresses(FBUtilities.getBroadcastAddressAndPort()), + NodeVersion.CURRENT)); + } + catch (Throwable t) + { + fail("Should not happen"); + } + }); + }); + commitStartupTranformation.start(); + Thread.sleep(10_000); + startNode2.start(); + + commitStartupTranformation.join(); + startNode2.join(); + } + } +}