consensus-bizur/src/main/java/com/mboysan/consensus/BizurRun.java
/*----------------------------------------------------------------------------------
     * Algorithm 2: Bucket Replication: Write
     * ----------------------------------------------------------------------------------*/
Algorithm 2: Bucket Replication: Write
Algorithm 2 : Bucket Replication : Write
private void write(Bucket bucket) throws BizurException {
        validateIAmTheLeader();
        int electId = bizurState.getElectId();
        int index = bucket.getIndex();
        bucket.setVerElectId(electId);
        bucket.incrementVerCounter();
        BucketView bucketView = bucket.createView();
        AtomicInteger ackCount = new AtomicInteger(0);
        forEachPeerParallel(peer -> {
            ReplicaWriteRequest request = new ReplicaWriteRequest(index, bucketView)
                    .setCorrelationId(correlationId)
                    .setSenderId(getNodeId())
                    .setReceiverId(peer.peerId);
            try {
                ReplicaWriteResponse response = getRPC().replicaWrite(request);
                if (response.isAcked()) {
                    ackCount.incrementAndGet();
                }
            } catch (IOException e) {
                LOGGER.error("{} - peer-{} IO exception for request={}, cause={}",
                        contextInfo(), peer.peerId, request, e.getMessage());
            }
        });
        if (!isMajorityAcked(ackCount.get())) {
            bizurState.setLeaderId(-1);  
            throw new BizurException(String.format("%s - write failed for bucket index=%d, electId=%d",
                    contextInfo(), index, electId));
        }
    }
private void write ( Bucket bucket ) throws BizurException { validateIAmTheLeader ( ) ; int electId = bizurState . getElectId ( ) ; int index = bucket . getIndex ( ) ; bucket . setVerElectId ( electId ) ; bucket . incrementVerCounter ( ) ; BucketView bucketView = bucket . createView ( ) ; AtomicInteger ackCount = new AtomicInteger ( 0 ) ; forEachPeerParallel ( peer -> { ReplicaWriteRequest request = new ReplicaWriteRequest ( index , bucketView ) . setCorrelationId ( correlationId ) . setSenderId ( getNodeId ( ) ) . setReceiverId ( peer . peerId ) ; try { ReplicaWriteResponse response = getRPC ( ) . replicaWrite ( request ) ; if ( response . isAcked ( ) ) { ackCount . incrementAndGet ( ) ; } } catch ( IOException e ) { LOGGER . error ( "{} - peer-{} IO exception for request={}, cause={}" , contextInfo ( ) , peer . peerId , request , e . getMessage ( ) ) ; } } ) ; if ( ! isMajorityAcked ( ackCount . get ( ) ) ) { bizurState . setLeaderId ( - 1 ) ; throw new BizurException ( String . format ( "%s - write failed for bucket index=%d, electId=%d" , contextInfo ( ) , index , electId ) ) ; } }
Algorithm 2: Bucket Replication: Write
Algorithm 2 : Bucket Replication : Write
// step down