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.0' into cassandra-3.11

This commit is contained in:
Aleksey Yeschenko 2021-11-12 14:32:16 +00:00
parent c922ee1471 58c878e3e0
commit 40f4fe7bfd
3 changed files with 31 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -11,6 +11,7 @@
* Reduce thread contention in CommitLogSegment and HintsBuffer (CASSANDRA-16072)
* Avoid sending CDC column if not enabled (CASSANDRA-16770)
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)

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

@ -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;
@ -239,7 +238,7 @@ public class BatchlogManager implements BatchlogManagerMBean
int positionInPage = 0;
ArrayList<ReplayingBatch> unfinishedBatches = new ArrayList<>(pageSize);
Set<InetAddress> hintedNodes = new HashSet<>();
Set<UUID> hintedNodes = new HashSet<>();
Set<UUID> replayedBatches = new HashSet<>();
// Sending out batches for replay without waiting for them, so that one stuck batch doesn't affect others
@ -275,16 +274,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<ReplayingBatch> batches, Set<InetAddress> 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)
@ -319,7 +320,7 @@ public class BatchlogManager implements BatchlogManagerMBean
this.replayedBytes = addMutations(version, serializedMutations);
}
public int replay(RateLimiter rateLimiter, Set<InetAddress> hintedNodes) throws IOException
public int replay(RateLimiter rateLimiter, Set<UUID> hintedNodes) throws IOException
{
logger.trace("Replaying batch {}", id);
@ -337,7 +338,7 @@ public class BatchlogManager implements BatchlogManagerMBean
return replayHandlers.size();
}
public void finish(Set<InetAddress> hintedNodes)
public void finish(Set<UUID> hintedNodes)
{
for (int i = 0; i < replayHandlers.size(); i++)
{
@ -385,7 +386,7 @@ public class BatchlogManager implements BatchlogManagerMBean
mutations.add(mutation);
}
private void writeHintsForUndeliveredEndpoints(int startFrom, Set<InetAddress> hintedNodes)
private void writeHintsForUndeliveredEndpoints(int startFrom, Set<UUID> hintedNodes)
{
int gcgs = gcgs(mutations);
@ -393,6 +394,7 @@ public class BatchlogManager implements BatchlogManagerMBean
if (TimeUnit.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);
@ -400,16 +402,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<ReplayWriteResponseHandler<Mutation>> sendReplays(List<Mutation> mutations,
long writtenAt,
Set<InetAddress> hintedNodes)
Set<UUID> hintedNodes)
{
List<ReplayWriteResponseHandler<Mutation>> handlers = new ArrayList<>(mutations.size());
for (Mutation mutation : mutations)
@ -429,7 +438,7 @@ public class BatchlogManager implements BatchlogManagerMBean
*/
private static ReplayWriteResponseHandler<Mutation> sendSingleReplayMutation(final Mutation mutation,
long writtenAt,
Set<InetAddress> hintedNodes)
Set<UUID> hintedNodes)
{
Set<InetAddress> liveEndpoints = new HashSet<>();
String ks = mutation.getKeyspaceName();
@ -447,9 +456,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);
}
}
}

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

@ -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;
@ -186,7 +187,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);
}