consensus-bizur/src/main/java/com/mboysan/consensus/BizurRun.java
/*----------------------------------------------------------------------------------
     * Algorithm 4: Bucket Replication: Recovery
     * ----------------------------------------------------------------------------------*/
Algorithm 4: Bucket Replication: Recovery
Algorithm 4 : Bucket Replication : Recovery
private void ensureRecovery(Bucket bucket, int electId) throws BizurException {
        if (electId == bucket.getVerElectId()) {
            return;
        }
        int index = bucket.getIndex();
        AtomicReference<BucketView> maxVerBucketView = new AtomicReference<>(null);
        AtomicInteger ackCount = new AtomicInteger(0);
        forEachPeerParallel(peer -> {
            ReplicaReadRequest request = new ReplicaReadRequest(index, electId)
                    .setCorrelationId(correlationId)
                    .setSenderId(getNodeId())
                    .setReceiverId(peer.peerId);
            try {
                ReplicaReadResponse response = getRPC().replicaRead(request);
                if (response.isAcked()) {
                    BucketView bucketView = response.getBucketView();
                    synchronized (maxVerBucketView) {
                        if (!maxVerBucketView.compareAndSet(null, bucketView)) {
                            if (bucketView.compareTo(maxVerBucketView.get()) > 0) {
                                maxVerBucketView.set(bucketView);
                            }
                        }
                    }
                    ackCount.incrementAndGet();
                }
            } catch (IOException e) {
                LOGGER.error("{} - peer-{} IO exception for request={}, cause={}",
                        contextInfo(), peer.peerId, request, e.getMessage());
            }
        });
        if (isMajorityAcked(ackCount.get())) {
            BucketView bucketView = maxVerBucketView.get();
            bucket.setVerElectId(electId)
                    .setVerCounter(0)
                    .setBucketMap(bucketView.getBucketMap());
            write(bucket);
        } else {
            bizurState.setLeaderId(-1);  
            throw new BizurException(String.format("%s - could not ensure recovery of bucket index=%d, electId=%d",
                    contextInfo(), index, electId));
        }
    }
private void ensureRecovery ( Bucket bucket , int electId ) throws BizurException { if ( electId == bucket . getVerElectId ( ) ) { return ; } int index = bucket . getIndex ( ) ; AtomicReference < BucketView > maxVerBucketView = new AtomicReference < > ( null ) ; AtomicInteger ackCount = new AtomicInteger ( 0 ) ; forEachPeerParallel ( peer -> { ReplicaReadRequest request = new ReplicaReadRequest ( index , electId ) . setCorrelationId ( correlationId ) . setSenderId ( getNodeId ( ) ) . setReceiverId ( peer . peerId ) ; try { ReplicaReadResponse response = getRPC ( ) . replicaRead ( request ) ; if ( response . isAcked ( ) ) { BucketView bucketView = response . getBucketView ( ) ; synchronized ( maxVerBucketView ) { if ( ! maxVerBucketView . compareAndSet ( null , bucketView ) ) { if ( bucketView . compareTo ( maxVerBucketView . get ( ) ) > 0 ) { maxVerBucketView . set ( bucketView ) ; } } } ackCount . incrementAndGet ( ) ; } } catch ( IOException e ) { LOGGER . error ( "{} - peer-{} IO exception for request={}, cause={}" , contextInfo ( ) , peer . peerId , request , e . getMessage ( ) ) ; } } ) ; if ( isMajorityAcked ( ackCount . get ( ) ) ) { BucketView bucketView = maxVerBucketView . get ( ) ; bucket . setVerElectId ( electId ) . setVerCounter ( 0 ) . setBucketMap ( bucketView . getBucketMap ( ) ) ; write ( bucket ) ; } else { bizurState . setLeaderId ( - 1 ) ; throw new BizurException ( String . format ( "%s - could not ensure recovery of bucket index=%d, electId=%d" , contextInfo ( ) , index , electId ) ) ; } }
Algorithm 4: Bucket Replication: Recovery
Algorithm 4 : Bucket Replication : Recovery
// step down