From 58c878e3e0356286ac1418e5e52e6f5cadd22ddb Mon Sep 17 00:00:00 2001 From: Aleksey Yeschenko Date: Tue, 19 Oct 2021 15:44:09 +0100 Subject: [PATCH] Fix rare NPE caused by batchlog replay / node decomission races pach by Aleksey Yeschenko; reviewed by Alex Petrov for CASSANDRA-17049 --- CHANGES.txt | 1 + .../cassandra/batchlog/BatchlogManager.java | 44 ++++++++++++------- .../apache/cassandra/hints/HintsService.java | 3 +- 3 files changed, 31 insertions(+), 17 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 768c45e53d..cf6a956c90 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.26: + * Fix rare NPE caused by batchlog replay / node decomission races (CASSANDRA-17049) * Allow users to view permissions of the roles they created (CASSANDRA-16902) * Fix failure handling in inter-node communication (CASSANDRA-16334) * Log more information when a node runs out of commitlog space (CASSANDRA-11323) diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java index 71d60e7df3..100c8602b5 100644 --- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java +++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java @@ -77,7 +77,6 @@ import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MBeanWrapper; import org.apache.cassandra.utils.UUIDGen; -import static com.google.common.collect.Iterables.transform; import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; import static org.apache.cassandra.cql3.QueryProcessor.executeInternalWithPaging; @@ -238,7 +237,7 @@ public class BatchlogManager implements BatchlogManagerMBean int positionInPage = 0; ArrayList unfinishedBatches = new ArrayList<>(pageSize); - Set hintedNodes = new HashSet<>(); + Set hintedNodes = new HashSet<>(); Set replayedBatches = new HashSet<>(); // Sending out batches for replay without waiting for them, so that one stuck batch doesn't affect others @@ -274,16 +273,18 @@ public class BatchlogManager implements BatchlogManagerMBean } } - finishAndClearBatches(unfinishedBatches, hintedNodes, replayedBatches); + // finalize the incomplete last page of batches + if (positionInPage > 0) + finishAndClearBatches(unfinishedBatches, hintedNodes, replayedBatches); // to preserve batch guarantees, we must ensure that hints (if any) have made it to disk, before deleting the batches - HintsService.instance.flushAndFsyncBlockingly(transform(hintedNodes, StorageService.instance::getHostIdForEndpoint)); + HintsService.instance.flushAndFsyncBlockingly(hintedNodes); // once all generated hints are fsynced, actually delete the batches replayedBatches.forEach(BatchlogManager::remove); } - private void finishAndClearBatches(ArrayList batches, Set hintedNodes, Set replayedBatches) + private void finishAndClearBatches(ArrayList batches, Set hintedNodes, Set replayedBatches) { // schedule hints for timed out deliveries for (ReplayingBatch batch : batches) @@ -318,7 +319,7 @@ public class BatchlogManager implements BatchlogManagerMBean this.replayedBytes = addMutations(version, serializedMutations); } - public int replay(RateLimiter rateLimiter, Set hintedNodes) throws IOException + public int replay(RateLimiter rateLimiter, Set hintedNodes) throws IOException { logger.trace("Replaying batch {}", id); @@ -336,7 +337,7 @@ public class BatchlogManager implements BatchlogManagerMBean return replayHandlers.size(); } - public void finish(Set hintedNodes) + public void finish(Set hintedNodes) { for (int i = 0; i < replayHandlers.size(); i++) { @@ -384,7 +385,7 @@ public class BatchlogManager implements BatchlogManagerMBean mutations.add(mutation); } - private void writeHintsForUndeliveredEndpoints(int startFrom, Set hintedNodes) + private void writeHintsForUndeliveredEndpoints(int startFrom, Set hintedNodes) { int gcgs = gcgs(mutations); @@ -392,6 +393,7 @@ public class BatchlogManager implements BatchlogManagerMBean if (TimeUnit.MILLISECONDS.toSeconds(writtenAt) + gcgs <= FBUtilities.nowInSeconds()) return; + Set nodesToHint = new HashSet<>(); for (int i = startFrom; i < replayHandlers.size(); i++) { ReplayWriteResponseHandler handler = replayHandlers.get(i); @@ -399,16 +401,23 @@ public class BatchlogManager implements BatchlogManagerMBean if (handler != null) { - hintedNodes.addAll(handler.undelivered); - HintsService.instance.write(transform(handler.undelivered, StorageService.instance::getHostIdForEndpoint), - Hint.create(undeliveredMutation, writtenAt)); + for (InetAddress address : handler.undelivered) + { + UUID hostId = StorageService.instance.getHostIdForEndpoint(address); + if (null != hostId) + nodesToHint.add(hostId); + } + if (!nodesToHint.isEmpty()) + HintsService.instance.write(nodesToHint, Hint.create(undeliveredMutation, writtenAt)); + hintedNodes.addAll(nodesToHint); + nodesToHint.clear(); } } } private static List> sendReplays(List mutations, long writtenAt, - Set hintedNodes) + Set hintedNodes) { List> handlers = new ArrayList<>(mutations.size()); for (Mutation mutation : mutations) @@ -428,7 +437,7 @@ public class BatchlogManager implements BatchlogManagerMBean */ private static ReplayWriteResponseHandler sendSingleReplayMutation(final Mutation mutation, long writtenAt, - Set hintedNodes) + Set hintedNodes) { Set liveEndpoints = new HashSet<>(); String ks = mutation.getKeyspaceName(); @@ -446,9 +455,12 @@ public class BatchlogManager implements BatchlogManagerMBean } else { - hintedNodes.add(endpoint); - HintsService.instance.write(StorageService.instance.getHostIdForEndpoint(endpoint), - Hint.create(mutation, writtenAt)); + UUID hostId = StorageService.instance.getHostIdForEndpoint(endpoint); + if (null != hostId) + { + HintsService.instance.write(hostId, Hint.create(mutation, writtenAt)); + hintedNodes.add(hostId); + } } } diff --git a/src/java/org/apache/cassandra/hints/HintsService.java b/src/java/org/apache/cassandra/hints/HintsService.java index 00d1954695..6546836880 100644 --- a/src/java/org/apache/cassandra/hints/HintsService.java +++ b/src/java/org/apache/cassandra/hints/HintsService.java @@ -21,6 +21,7 @@ import java.io.File; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.Collections; +import java.util.Objects; import java.util.UUID; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; @@ -187,7 +188,7 @@ public final class HintsService implements HintsServiceMBean */ public void flushAndFsyncBlockingly(Iterable hostIds) { - Iterable stores = transform(hostIds, catalog::get); + Iterable stores = filter(transform(hostIds, catalog::getNullable), Objects::nonNull); writeExecutor.flushBufferPool(bufferPool, stores); writeExecutor.fsyncWritersBlockingly(stores); }