natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.11' into cassandra-4.0

This commit is contained in:
Aleksey Yeschenko 2021-11-12 14:40:33 +00:00
parent dd6242037b 40f4fe7bfd
commit 03e83f2070
3 changed files with 33 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -22,6 +22,7 @@ Merged from 3.11:
* Update Jackson from 2.9.10 to 2.12.5 (CASSANDRA-16851)
* Make assassinate more resilient to missing tokens (CASSANDRA-16847)
Merged from 3.0:
* 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)

46
src/java/org/apache/cassandra/batchlog/BatchlogManager.java
View File

@ -35,7 +35,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Collections2;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
@ -78,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 java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternalWithPaging;
@ -264,7 +262,7 @@ public class BatchlogManager implements BatchlogManagerMBean
int positionInPage = 0;
ArrayList<ReplayingBatch> unfinishedBatches = new ArrayList<>(pageSize);
Set<InetAddressAndPort> hintedNodes = new HashSet<>();
Set<UUID> hintedNodes = new HashSet<>();
Set<UUID> replayedBatches = new HashSet<>();
Exception caughtException = null;
int skipped = 0;
@ -304,19 +302,21 @@ public class BatchlogManager implements BatchlogManagerMBean
}
}
finishAndClearBatches(unfinishedBatches, hintedNodes, replayedBatches);
// finalize the incomplete last page of batches
if (positionInPage > 0)
finishAndClearBatches(unfinishedBatches, hintedNodes, replayedBatches);
if (caughtException != null)
logger.warn(String.format("Encountered %d unexpected exceptions while sending out batches", skipped), caughtException);
// 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<ReplayingBatch> batches, Set<InetAddressAndPort> hintedNodes, Set<UUID> replayedBatches)
private void finishAndClearBatches(ArrayList<ReplayingBatch> batches, Set<UUID> hintedNodes, Set<UUID> replayedBatches)
{
// schedule hints for timed out deliveries
for (ReplayingBatch batch : batches)
@ -351,7 +351,7 @@ public class BatchlogManager implements BatchlogManagerMBean
this.replayedBytes = addMutations(version, serializedMutations);
}
public int replay(RateLimiter rateLimiter, Set<InetAddressAndPort> hintedNodes) throws IOException
public int replay(RateLimiter rateLimiter, Set<UUID> hintedNodes) throws IOException
{
logger.trace("Replaying batch {}", id);
@ -369,7 +369,7 @@ public class BatchlogManager implements BatchlogManagerMBean
return replayHandlers.size();
}
public void finish(Set<InetAddressAndPort> hintedNodes)
public void finish(Set<UUID> hintedNodes)
{
for (int i = 0; i < replayHandlers.size(); i++)
{
@ -417,7 +417,7 @@ public class BatchlogManager implements BatchlogManagerMBean
mutations.add(mutation);
}
private void writeHintsForUndeliveredEndpoints(int startFrom, Set<InetAddressAndPort> hintedNodes)
private void writeHintsForUndeliveredEndpoints(int startFrom, Set<UUID> hintedNodes)
{
int gcgs = gcgs(mutations);
@ -425,6 +425,7 @@ public class BatchlogManager implements BatchlogManagerMBean
if (MILLISECONDS.toSeconds(writtenAt) + gcgs <= FBUtilities.nowInSeconds())
return;
Set<UUID> nodesToHint = new HashSet<>();
for (int i = startFrom; i < replayHandlers.size(); i++)
{
ReplayWriteResponseHandler<Mutation> handler = replayHandlers.get(i);
@ -432,16 +433,23 @@ public class BatchlogManager implements BatchlogManagerMBean
if (handler != null)
{
hintedNodes.addAll(handler.undelivered);
HintsService.instance.write(Collections2.transform(handler.undelivered, StorageService.instance::getHostIdForEndpoint),
Hint.create(undeliveredMutation, writtenAt));
for (InetAddressAndPort 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<ReplayWriteResponseHandler<Mutation>> sendReplays(List<Mutation> mutations,
long writtenAt,
Set<InetAddressAndPort> hintedNodes)
Set<UUID> hintedNodes)
{
List<ReplayWriteResponseHandler<Mutation>> handlers = new ArrayList<>(mutations.size());
for (Mutation mutation : mutations)
@ -460,7 +468,7 @@ public class BatchlogManager implements BatchlogManagerMBean
*/
private static ReplayWriteResponseHandler<Mutation> sendSingleReplayMutation(final Mutation mutation,
long writtenAt,
Set<InetAddressAndPort> hintedNodes)
Set<UUID> hintedNodes)
{
String ks = mutation.getKeyspaceName();
Keyspace keyspace = Keyspace.open(ks);
@ -484,9 +492,13 @@ public class BatchlogManager implements BatchlogManagerMBean
{
if (replica == selfReplica || liveRemoteOnly.all().contains(replica))
continue;
hintedNodes.add(replica.endpoint());
HintsService.instance.write(StorageService.instance.getHostIdForEndpoint(replica.endpoint()),
Hint.create(mutation, writtenAt));
UUID hostId = StorageService.instance.getHostIdForEndpoint(replica.endpoint());
if (null != hostId)
{
HintsService.instance.write(hostId, Hint.create(mutation, writtenAt));
hintedNodes.add(hostId);
}
}
ReplicaPlan.ForTokenWrite replicaPlan = new ReplicaPlan.ForTokenWrite(keyspace, liveAndDown.replicationStrategy(),

4
src/java/org/apache/cassandra/hints/HintsService.java
View File

@ -22,6 +22,7 @@ import java.net.UnknownHostException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
@ -49,6 +50,7 @@ import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.MBeanWrapper;
import static com.google.common.collect.Iterables.filter;
import static com.google.common.collect.Iterables.transform;
/**
@ -195,7 +197,7 @@ public final class HintsService implements HintsServiceMBean
*/
public void flushAndFsyncBlockingly(Iterable<UUID> hostIds)
{
Iterable<HintsStore> stores = transform(hostIds, catalog::get);
Iterable<HintsStore> stores = filter(transform(hostIds, catalog::getNullable), Objects::nonNull);
writeExecutor.flushBufferPool(bufferPool, stores);
writeExecutor.fsyncWritersBlockingly(stores);
}