ZOOKEEPER-5088: Initialize ObserverMaster with pending follower proposals

[JHSUYU]

JIRA: ZOOKEEPER-5088

Description

A follower acting as an ObserverMaster can receive an uncommitted proposal while synchronizing with the leader.

The relevant lifecycle in Follower.followLeader() is:

syncWithLeader(newEpochZxid);

if (self.getObserverMasterPort() > 0) {
    om = new ObserverMaster(self, fzk, self.getObserverMasterPort());
    om.start();
}

During syncWithLeader(), outstanding proposals are converted to requests and logged directly:

for (PacketInFlight p : packetsNotLogged) {
    fzk.logRequest(p.toRequest());
}

FollowerZooKeeperServer.logRequest() adds these proposals to the follower's pending transaction queue:

if (request.getHdr() != null) {
    pendingTxns.add(request);
}

However, ObserverMaster is created only after synchronization completes. These proposals therefore do not pass through the normal Follower.processPacket(PROPOSAL) path that populates both queues:

fzk.logRequest(logEntry.toRequest());

if (om != null) {
    om.proposalReceived(qp);
}

This can leave the follower and its ObserverMaster with different views of the same in-flight transaction:

FollowerZooKeeperServer.pendingTxns = [Z3]
ObserverMaster.proposedPkts         = []

If Z3 remains uncommitted until after ObserverMaster starts, the later COMMIT(Z3) reaches:

fzk.commit(qp.getZxid());

if (om != null) {
    om.proposalCommitted(qp.getZxid());
}

The follower finds Z3 in pendingTxns and applies it locally. ObserverMaster.proposalCommitted(Z3), however, cannot find Z3 in proposedPkts:

QuorumPacket pkt = proposedPkts.peek();
if (pkt == null || pkt.getZxid() > zxid) {
    return null;
}

It consequently returns without sending INFORM(Z3) to its downstream observers.

The next proposal, Z4, follows the normal packet-processing path after ObserverMaster exists. It is added to proposedPkts and forwarded successfully, causing the observer to receive Z4 directly after Z2:

follower applied:  Z2 -> Z3 -> Z4
observer received: Z2 ------> Z4

When the observer processes Z4, it reaches the continuity check in ZKDatabase.addCommittedProposal():

} else if (!allowDiscontinuousProposals
        && request.zxid != maxCommittedLog + 1
        && ZxidUtils.getEpochFromZxid(request.zxid)
            <= ZxidUtils.getEpochFromZxid(maxCommittedLog)) {
    throw new IllegalStateException(...);
}

Because the observer's maxCommittedLog is Z2 while the incoming transaction is Z4, it throws:

Committed proposal cached out of order:
0x100000004 is not the next proposal of 0x100000002

This is treated as a severe unrecoverable error and terminates the observer's critical CommitProcessor thread.

Solution

Initialize the new ObserverMaster with the follower's existing pendingTxns before it starts accepting observer connections.

Each pending Request is serialized into the same INFORM representation used by the normal proposal path and added to ObserverMaster.proposedPkts.

This preserves the handoff between follower synchronization and normal broadcasting:

pendingTxns contains Z
    -> proposedPkts contains Z
    -> COMMIT(Z) sends INFORM(Z)

Tests

Added ObserverMasterPendingProposalTest, which runs three voting participants and one observer in a single JVM.

The test:

1. Restarts the follower serving as the observer's ObserverMaster.
2. Creates an in-flight proposal while the follower is synchronizing.
3. Delays its ACKs until synchronization and observer reconnection complete.
4. Commits the in-flight proposal followed by another proposal.
5. Verifies that the observer receives both transactions in order and that its CommitProcessor remains running.

Without the fix, the observer skips the first proposal, receives an out-of-order zxid, throws IllegalStateException, and terminates its CommitProcessor.

[JHSUYU]

Hi @kezhuw @anmolnar, could you take a look at this PR when you have a chance? Thanks!

[JHSUYU]

The unit test is somewhat tricky and ugly. Happy to revise if you have better suggestions.