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

This commit is contained in:
Stefan Miklosovic 2023-11-13 15:18:25 +01:00
parent a5299bb453 fe1be800b4
commit afb422ddc0
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
6 changed files with 36 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -6,6 +6,7 @@
* Allow empty keystore_password in encryption_options (CASSANDRA-18778)
* Skip ColumnFamilyStore#topPartitions initialization when client or tool mode (CASSANDRA-18697)
Merged from 4.0:
* Remove completed coordinator sessions (CASSANDRA-18903)
* Make StartupConnectivityChecker only run a connectivity check if there are no nodes which are running a version prior to Cassandra 4 (CASSANDRA-18968)
* Retrieve keyspaces metadata and schema version concistently in DescribeStatement (CASSANDRA-18921)
* Gossip NPE due to shutdown event corrupting empty statuses (CASSANDRA-18913)

6
src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
View File

@ -206,6 +206,12 @@ public abstract class ConsistentSession
this.participants = ImmutableSet.copyOf(builder.participants);
}
public boolean isCompleted()
{
State s = getState();
return s == State.FINALIZED || s == State.FAILED;
}
public State getState()
{
return state;

13
src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.repair.consistent;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@ -70,9 +71,12 @@ public class CoordinatorSession extends ConsistentSession
private volatile long repairStart = Long.MIN_VALUE;
private volatile long finalizeStart = Long.MIN_VALUE;
private final Consumer<CoordinatorSession> listener;
public CoordinatorSession(Builder builder)
{
super(builder);
this.listener = builder.listener;
for (InetAddressAndPort participant : participants)
{
participantStates.put(participant, State.PREPARING);
@ -81,6 +85,13 @@ public class CoordinatorSession extends ConsistentSession
public static class Builder extends AbstractBuilder
{
Consumer<CoordinatorSession> listener;
public void withListener(Consumer<CoordinatorSession> listener)
{
this.listener = listener;
}
public CoordinatorSession build()
{
validate();
@ -97,6 +108,8 @@ public class CoordinatorSession extends ConsistentSession
{
logger.trace("Setting coordinator state to {} for repair {}", state, sessionID);
super.setState(state);
if (listener != null)
listener.accept(this);
}
@VisibleForTesting

14
src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java
View File

@ -24,6 +24,9 @@ import java.util.Set;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.repair.messages.FailSession;
import org.apache.cassandra.repair.messages.FinalizePromise;
@ -37,6 +40,7 @@ import org.apache.cassandra.utils.TimeUUID;
*/
public class CoordinatorSessions
{
private static final Logger logger = LoggerFactory.getLogger(CoordinatorSessions.class);
private final Map<TimeUUID, CoordinatorSession> sessions = new HashMap<>();
protected CoordinatorSession buildSession(CoordinatorSession.Builder builder)
@ -62,6 +66,7 @@ public class CoordinatorSessions
builder.withRepairedAt(prs.repairedAt);
builder.withRanges(prs.getRanges());
builder.withParticipants(participants);
builder.withListener(this::onSessionStateUpdate);
CoordinatorSession session = buildSession(builder);
sessions.put(session.sessionID, session);
return session;
@ -72,6 +77,15 @@ public class CoordinatorSessions
return sessions.get(sessionId);
}
public synchronized void onSessionStateUpdate(CoordinatorSession session)
{
if (session.isCompleted())
{
logger.info("Removing completed session {} with state {}", session.sessionID, session.getState());
sessions.remove(session.sessionID);
}
}
public void handlePrepareResponse(PrepareConsistentResponse msg)
{
CoordinatorSession session = getSession(msg.parentSession);

6
src/java/org/apache/cassandra/repair/consistent/LocalSession.java
View File

@ -37,12 +37,6 @@ public class LocalSession extends ConsistentSession
this.lastUpdate = builder.lastUpdate;
}
public boolean isCompleted()
{
State s = getState();
return s == State.FINALIZED || s == State.FAILED;
}
public int getStartedAt()
{
return startedAt;

3
test/distributed/org/apache/cassandra/distributed/test/IncRepairCoordinatorErrorTest.java
View File

@ -46,11 +46,12 @@ public class IncRepairCoordinatorErrorTest extends TestBaseImpl
.to(3)
.messagesMatching((from, to, msg) -> msg.verb() == FINALIZE_COMMIT_MSG.id).drop();
cluster.get(1).nodetoolResult("repair", KEYSPACE).asserts().success();
assertThat(cluster.get(1).logs().watchFor("Removing completed session .* with state FINALIZED").getResult()).isNotEmpty();
TimeUUID result = (TimeUUID) cluster.get(1).executeInternal("select parent_id from system_distributed.repair_history")[0][0];
cluster.get(3).runOnInstance(() -> {
ActiveRepairService.instance.failSession(result.toString(), true);
});
assertThat(cluster.get(1).logs().watchFor("Can't transition endpoints .* to FAILED").getResult()).isNotEmpty();
}
}
}