alluxio.master.journal.raft

Class RaftJournalSystem

java.lang.Object

alluxio.master.journal.AbstractJournalSystem

alluxio.master.journal.raft.RaftJournalSystem

All Implemented Interfaces:

JournalSystem

@ThreadSafe
public class RaftJournalSystem
extends AbstractJournalSystem

System for multiplexing many logical journals into a single raft-based journal. This class embeds a RaftServer which implements the raft algorithm, replicating entries across a majority of servers before applying them to the state machine. To make the Ratis system work as an Alluxio journal system, we implement two non-standard behaviors: (1) pre-applying operations on the primary and (2) tightly controlling primary snapshotting.

Pre-apply

Unlike the Ratis framework, Alluxio updates state machine state *before* writing to the journal. This lets us avoid journaling operations which do not result in state modification. To make this work in the Ratis framework, we allow RPCs to modify state directly, then write an entry to Ratis afterwards. Once the entry is journaled, Ratis will attempt to apply the journal entry to each master. The entry has already been applied on the primary, so we treat all journal entries applied to the primary as no-ops to avoid double-application.

Correctness of pre-apply

There are two cases to worry about: (1) incorrectly ignoring an entry and (2) incorrectly applying an entry.

Avoid incorrectly ignoring entries

This could happen if a server thinks it is the primary, and ignores a journal entry served from the real primary. To prevent this, primaries wait for a quiet period before serving requests. During this time, the previous primary will go through at least two election cycles without successfully sending heartbeats to the majority of the cluster. If the old primary successfully sent a heartbeat to a node which elected the new primary, the old primary would realize it isn't primary and step down. Therefore, the old primary will step down and no longer ignore requests by the time the new primary begins sending entries.

Avoid incorrectly applying entries

Entries can never be double-applied to a primary's state because as long as it is the primary, it will ignore all entries, and once it becomes standby, it will completely reset its state and rejoin the cluster.

Snapshot control

The way we apply journal entries to the primary makes it tricky to perform primary state snapshots. Normally Ratis would decide when it wants a snapshot, but with the pre-apply protocol we may be in the middle of modifying state when the snapshot would happen. To manage this, we declare an AtomicBoolean field which decides whether it will be allowed to take snapshots. Normally, snapshots are prohibited on the primary. However, we don't want the primary's log to grow unbounded, so we allow a snapshot to be taken once a day at a user-configured time. To support this, all state changes must first acquire a read lock, and snapshotting requires the corresponding write lock. Once we have the write lock for all state machines, we enable snapshots in Ratis through our AtomicBoolean, then wait for any snapshot to complete.

Nested Class Summary

Nested classes/interfaces inherited from interface alluxio.master.journal.JournalSystem

JournalSystem.Builder, JournalSystem.Mode

Field Summary

Fields

Modifier and Type	Field and Description
static org.apache.ratis.protocol.RaftGroupId	RAFT_GROUP_ID
static UUID	RAFT_GROUP_UUID

Constructor Summary

Constructors

Constructor and Description
RaftJournalSystem(URI path, NetworkAddressUtils.ServiceType serviceType) Creates a RaftJournalSystem.

Method Summary

Modifier and Type	Method and Description
void	addQuorumServer(NetAddress serverNetAddress) Adds a server to the quorum.
CatchupFuture	catchup(Map<String,Long> journalSequenceNumbers) Initiates a catching up of journals to given sequences.
void	checkpoint(StateLockManager stateLockManager) Creates a checkpoint in the primary master journal system.
Journal	createJournal(Master master) Creates a journal for the given state machine.
void	format() Formats the journal system.
void	gainPrimacy() Transitions the journal to primary mode.
org.apache.ratis.protocol.RaftGroup	getCurrentGroup()
Map<String,Long>	getCurrentSequenceNumbers() Used to get the current state from a leader journal system.
Map<ServiceType,GrpcService>	getJournalServices()
String	getLeaderId() Get the leader id.
protected int	getLeaderIndex() Gets leader index.
org.apache.ratis.protocol.RaftPeerId	getLocalPeerId()
PrimarySelector	getPrimarySelector()
List<QuorumServerInfo>	getQuorumServerInfoList() Used to get information of internal RAFT quorum.
int	getRoleId() Get the role index.
boolean	isEmpty() Returns whether the journal is formatted and has not had any entries written to it yet.
boolean	isFormatted()
boolean	isLeader()
boolean	isSuspended()
void	losePrimacy() Transitions the journal to standby mode.
void	notifyLeadershipStateChanged(boolean isLeader) Notifies the journal that the leadership state has changed.
void	removeQuorumServer(NetAddress serverNetAddress) Removes from RAFT quorum, a server with given address.
void	resetPriorities() Resets RaftPeer priorities.
void	resume() Resumes applying for all journals.
CompletableFuture<org.apache.ratis.protocol.RaftClientReply>	sendMessageAsync(org.apache.ratis.protocol.RaftPeerId server, org.apache.ratis.protocol.Message message) Sends a message to a raft server asynchronously.
CompletableFuture<org.apache.ratis.protocol.RaftClientReply>	sendMessageAsync(org.apache.ratis.protocol.RaftPeerId server, org.apache.ratis.protocol.Message message, long timeoutMs) Sends a message to a raft server asynchronously.
void	startInternal() Throws AlluxioRuntimeException when it cannot start a RaftCluster and therefore cannot join the quorum.
void	stopInternal() Stops the journal system.
void	suspend(Runnable interruptCallback) Suspends applying for all journals.
String	toString()
String	transferLeadership(NetAddress newLeaderNetAddress) Transfers the leadership of the quorum to another server.
void	updateGroup() Updates raft group with the current values from raft server.

Methods inherited from class alluxio.master.journal.AbstractJournalSystem

addJournalSink, getJournalSinks, registerMetrics, removeJournalSink, start, stop

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface alluxio.master.journal.JournalSystem

waitForCatchup

Field Detail

RAFT_GROUP_UUID

public static final UUID RAFT_GROUP_UUID

RAFT_GROUP_ID

public static final org.apache.ratis.protocol.RaftGroupId RAFT_GROUP_ID

Constructor Detail

RaftJournalSystem

public RaftJournalSystem(URI path,
                         NetworkAddressUtils.ServiceType serviceType)

Creates a RaftJournalSystem.

Parameters:

path - where the journal will be stored

serviceType - is either MASTER_RAFT or JOB_MASTER_RAFT

Method Detail

getLocalPeerId

public org.apache.ratis.protocol.RaftPeerId getLocalPeerId()

Returns:

a raft peer id for local raft server

getCurrentGroup

public org.apache.ratis.protocol.RaftGroup getCurrentGroup()

Returns:

current raft group

createJournal

public Journal createJournal(Master master)

Description copied from interface: JournalSystem

Creates a journal for the given state machine. The returned journal can create journal contexts for writing journal entries. However, no entries may be written until the journal system has been started, and entries may only be written when the journal system is in PRIMARY mode. When the journal is started in standby mode, it will call Journaled#processJournalEntry(JournalEntry) and Journaled.resetState() to keep the state machine's state in sync with the entries written to the journal.

Parameters:

master - the master to create the journal for

Returns:

a new instance of Journal

gainPrimacy

public void gainPrimacy()

Description copied from interface: JournalSystem

Transitions the journal to primary mode.

losePrimacy

public void losePrimacy()

Description copied from interface: JournalSystem

Transitions the journal to standby mode.

getCurrentSequenceNumbers

public Map<String,Long> getCurrentSequenceNumbers()

Description copied from interface: JournalSystem

Used to get the current state from a leader journal system. Note: State changes to journals must have been effectively blocked by a state-lock before calling this method.

Returns:

the current map of sequences for each master

suspend

public void suspend(Runnable interruptCallback)
             throws IOException

Description copied from interface: JournalSystem

Suspends applying for all journals. When using suspend, the caller needs to provide a callback method as parameter. This callback is invoked when the journal needs to reload and thus cannot suspend the state changes any more. The callback should cancel any tasks that access the master states. After the callback returns, the journal assumes that the states is no longer being accessed and will reload immediately.

Parameters:

interruptCallback - the callback function to be invoked when the suspension is interrupted

Throws:

IOException

resume

public void resume()
            throws IOException

Description copied from interface: JournalSystem

Resumes applying for all journals. Note: Journal system should have been suspended prior to calling this.

Throws:

IOException

isSuspended

public boolean isSuspended()

Returns:

whether the journal is suspended

catchup

public CatchupFuture catchup(Map<String,Long> journalSequenceNumbers)

Description copied from interface: JournalSystem

Initiates a catching up of journals to given sequences. Note: Journal system should have been suspended prior to calling this.

Parameters:

journalSequenceNumbers - sequence to advance per each journal

Returns:

the future to track when catching up is completed

checkpoint

public void checkpoint(StateLockManager stateLockManager)
                throws IOException

Description copied from interface: JournalSystem

Creates a checkpoint in the primary master journal system.

Parameters:

stateLockManager - used to prevent reads and writes while the journal system is checkpointing

Throws:

IOException

getJournalServices

public Map<ServiceType,GrpcService> getJournalServices()

Returns:

RPC services for journal system

startInternal

public void startInternal()

Throws AlluxioRuntimeException when it cannot start a RaftCluster and therefore cannot join the quorum.

Specified by:

startInternal in class AbstractJournalSystem

stopInternal

public void stopInternal()

Description copied from class: AbstractJournalSystem

Stops the journal system.

Specified by:

stopInternal in class AbstractJournalSystem

getQuorumServerInfoList

public List<QuorumServerInfo> getQuorumServerInfoList()
                                               throws IOException

Used to get information of internal RAFT quorum.

Returns:

list of information for participating servers in RAFT quorum

Throws:

IOException

sendMessageAsync

public CompletableFuture<org.apache.ratis.protocol.RaftClientReply> sendMessageAsync(org.apache.ratis.protocol.RaftPeerId server,
                                                                                     org.apache.ratis.protocol.Message message)

Sends a message to a raft server asynchronously.

Parameters:

server - the raft peer id of the target server

message - the message to send

Returns:

a future to be completed with the client reply

sendMessageAsync

public CompletableFuture<org.apache.ratis.protocol.RaftClientReply> sendMessageAsync(org.apache.ratis.protocol.RaftPeerId server,
                                                                                     org.apache.ratis.protocol.Message message,
                                                                                     long timeoutMs)

Sends a message to a raft server asynchronously.

Parameters:

server - the raft peer id of the target server

message - the message to send

timeoutMs - the message timeout in milliseconds

Returns:

a future to be completed with the client reply

isLeader

public boolean isLeader()

Returns:

true if this journal system is the leader

removeQuorumServer

public void removeQuorumServer(NetAddress serverNetAddress)
                        throws IOException

Removes from RAFT quorum, a server with given address. For server to be removed, it should be in unavailable state in quorum.

Parameters:

serverNetAddress - address of the server to remove from the quorum

Throws:

IOException - raft exception

resetPriorities

public void resetPriorities()
                     throws IOException

Resets RaftPeer priorities.

Throws:

IOException - raft exception

transferLeadership

public String transferLeadership(NetAddress newLeaderNetAddress)

Transfers the leadership of the quorum to another server.

Parameters:

newLeaderNetAddress - the address of the server

Returns:

error message if an error occurs or empty string if no error occurred

addQuorumServer

public void addQuorumServer(NetAddress serverNetAddress)
                     throws IOException

Adds a server to the quorum.

Parameters:

serverNetAddress - the address of the server

Throws:

IOException - if error occurred while performing the operation

isEmpty

public boolean isEmpty()

Description copied from interface: JournalSystem

Returns whether the journal is formatted and has not had any entries written to it yet. This can only be determined when the journal system is in primary mode because entries are written to the primary first.

Returns:

whether the journal system is freshly formatted

isFormatted

public boolean isFormatted()

Returns:

whether the journal system has been formatted

format

public void format()
            throws IOException

Description copied from interface: JournalSystem

Formats the journal system.

Throws:

IOException

toString

public String toString()

Overrides:

toString in class Object

getPrimarySelector

public PrimarySelector getPrimarySelector()

Returns:

a primary selector that reflects the Raft quorum status

notifyLeadershipStateChanged

public void notifyLeadershipStateChanged(boolean isLeader)

Notifies the journal that the leadership state has changed.

Parameters:

isLeader - whether the local server is teh current leader

updateGroup

public void updateGroup()

Updates raft group with the current values from raft server.

getRoleId

public int getRoleId()

Get the role index. RaftProtos.RaftPeerRole.

Returns:

the role enum

getLeaderId

public String getLeaderId()

Get the leader id. RaftProtos.RaftPeerRole.

Returns:

the leader id

getLeaderIndex

protected int getLeaderIndex()

Gets leader index. The return integer means the leader index of embedded journal addresses -1 means leader not found.

Returns:

the leader index