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

Merge branch 'cassandra-4.1' into trunk

This commit is contained in:
Stefan Miklosovic 2023-06-15 13:23:47 +02:00
parent a55d4183f5 2b98d487d0
commit c7260399d6
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
3 changed files with 65 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -154,6 +154,7 @@ Merged from 4.0:
* Fix Down nodes counter in nodetool describecluster (CASSANDRA-18512)
* Remove unnecessary shuffling of GossipDigests in Gossiper#makeRandomGossipDigest (CASSANDRA-18546)
Merged from 3.11:
* Wait for live endpoints in gossip waiting to settle (CASSANDRA-18543)
* Fix error message handling when trying to use CLUSTERING ORDER with non-clustering column (CASSANDRA-17818
* Add keyspace and table name to exception message during ColumnSubselection deserialization (CASSANDRA-18346)
Merged from 3.0:

16
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -234,6 +234,22 @@ public enum CassandraRelevantProperties
GOSSIPER_QUARANTINE_DELAY("cassandra.gossip_quarantine_delay_ms"),
GOSSIPER_SKIP_WAITING_TO_SETTLE("cassandra.skip_wait_for_gossip_to_settle", "-1"),
GOSSIP_DISABLE_THREAD_VALIDATION("cassandra.gossip.disable_thread_validation"),
/**
* Delay before checking if gossip is settled.
*/
GOSSIP_SETTLE_MIN_WAIT_MS("cassandra.gossip_settle_min_wait_ms", "5000"),
/**
* Interval delay between checking gossip is settled.
*/
GOSSIP_SETTLE_POLL_INTERVAL_MS("cassandra.gossip_settle_interval_ms", "1000"),
/**
* Number of polls without gossip state change to consider gossip as settled.
*/
GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED("cassandra.gossip_settle_poll_success_required", "3"),
IGNORED_SCHEMA_CHECK_ENDPOINTS("cassandra.skip_schema_check_for_endpoints"),
IGNORED_SCHEMA_CHECK_VERSIONS("cassandra.skip_schema_check_for_versions"),
IGNORE_CORRUPTED_SCHEMA_TABLES("cassandra.ignore_corrupted_schema_tables"),

55
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -61,9 +61,11 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.FutureTask;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
@ -160,6 +162,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
@VisibleForTesting
final Set<InetAddressAndPort> liveEndpoints = new ConcurrentSkipListSet<>();
/* Inflight echo requests. */
private final Set<InetAddressAndPort> inflightEcho = new ConcurrentSkipListSet<>();
/* unreachable member set */
private final Map<InetAddressAndPort, Long> unreachableEndpoints = new ConcurrentHashMap<>();
@ -671,6 +676,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
return;
liveEndpoints.remove(endpoint);
inflightEcho.remove(endpoint);
unreachableEndpoints.remove(endpoint);
MessagingService.instance().versions.reset(endpoint);
quarantineEndpoint(endpoint);
@ -1361,14 +1367,45 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private void markAlive(final InetAddressAndPort addr, final EndpointState localState)
{
if (inflightEcho.contains(addr))
{
return;
}
inflightEcho.add(addr);
localState.markDead();
Message<NoPayload> echoMessage = Message.out(ECHO_REQ, noPayload);
logger.trace("Sending ECHO_REQ to {}", addr);
RequestCallback echoHandler = msg ->
RequestCallback echoHandler = new RequestCallback()
{
// force processing of the echo response onto the gossip stage, as it comes in on the REQUEST_RESPONSE stage
runInGossipStageBlocking(() -> realMarkAlive(addr, localState));
@Override
public void onResponse(Message msg)
{
// force processing of the echo response onto the gossip stage, as it comes in on the REQUEST_RESPONSE stage
runInGossipStageBlocking(() -> {
try
{
realMarkAlive(addr, localState);
}
finally
{
inflightEcho.remove(addr);
}
});
}
@Override
public boolean invokeOnFailure()
{
return true;
}
@Override
public void onFailure(InetAddressAndPort from, RequestFailureReason failureReason)
{
inflightEcho.remove(addr);
}
};
MessagingService.instance().sendWithCallback(echoMessage, addr, echoHandler);
@ -1424,6 +1461,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
if (!disableEndpointRemoval)
{
liveEndpoints.remove(addr);
inflightEcho.remove(addr);
unreachableEndpoints.put(addr, nanoTime());
}
}
@ -2378,21 +2416,23 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
return;
}
final int GOSSIP_SETTLE_MIN_WAIT_MS = 5000;
final int GOSSIP_SETTLE_POLL_INTERVAL_MS = 1000;
final int GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED = 3;
final int GOSSIP_SETTLE_MIN_WAIT_MS = CassandraRelevantProperties.GOSSIP_SETTLE_MIN_WAIT_MS.getInt();
final int GOSSIP_SETTLE_POLL_INTERVAL_MS = CassandraRelevantProperties.GOSSIP_SETTLE_POLL_INTERVAL_MS.getInt();
final int GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED = CassandraRelevantProperties.GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED.getInt();
logger.info("Waiting for gossip to settle...");
Uninterruptibles.sleepUninterruptibly(GOSSIP_SETTLE_MIN_WAIT_MS, TimeUnit.MILLISECONDS);
int totalPolls = 0;
int numOkay = 0;
int epSize = Gossiper.instance.getEndpointCount();
int liveSize = Gossiper.instance.getLiveMembers().size();
while (numOkay < GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED)
{
Uninterruptibles.sleepUninterruptibly(GOSSIP_SETTLE_POLL_INTERVAL_MS, TimeUnit.MILLISECONDS);
int currentSize = Gossiper.instance.getEndpointCount();
int currentLive = Gossiper.instance.getLiveMembers().size();
totalPolls++;
if (currentSize == epSize)
if (currentSize == epSize && currentLive == liveSize)
{
logger.debug("Gossip looks settled.");
numOkay++;
@ -2403,6 +2443,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
numOkay = 0;
}
epSize = currentSize;
liveSize = currentLive;
if (forceAfter > 0 && totalPolls > forceAfter)
{
logger.warn("Gossip not settled but startup forced by cassandra.skip_wait_for_gossip_to_settle. Gossip total polls: {}",