Feature: auto recover support repaired not adhering placement ledger

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java

@@ -567,7 +567,7 @@ public BookKeeper(ClientConfiguration conf, ZooKeeper zk, EventLoopGroup eventLo
         bookieQuarantineRatio = 1.0;
     }
 
-    private EnsemblePlacementPolicy initializeEnsemblePlacementPolicy(ClientConfiguration conf,
+    protected EnsemblePlacementPolicy initializeEnsemblePlacementPolicy(ClientConfiguration conf,
                                                                       DNSToSwitchMapping dnsResolver,
                                                                       HashedWheelTimer timer,
                                                                       FeatureProvider featureProvider,

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java

@@ -90,6 +90,7 @@
 import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.bookkeeper.util.AvailabilityOfEntriesOfLedger;
 import org.apache.bookkeeper.util.IOUtils;
+import org.apache.commons.collections4.MapUtils;
 import org.apache.zookeeper.AsyncCallback;
 import org.apache.zookeeper.KeeperException;
 import org.slf4j.Logger;
@@ -1129,14 +1130,24 @@ private ArrayList<BookieId> replaceBookiesInEnsemble(
      * @param ledgerFragment
      *            - LedgerFragment to replicate
      */
-    public void replicateLedgerFragment(LedgerHandle lh,
-            final LedgerFragment ledgerFragment,
-            final BiConsumer<Long, Long> onReadEntryFailureCallback)
-            throws InterruptedException, BKException {
-        Optional<Set<BookieId>> excludedBookies = Optional.empty();
-        Map<Integer, BookieId> targetBookieAddresses =
-                getReplacementBookiesByIndexes(lh, ledgerFragment.getEnsemble(),
-                        ledgerFragment.getBookiesIndexes(), excludedBookies);
+    public void replicateLedgerFragment(LedgerHandle lh, final LedgerFragment ledgerFragment,
+            final BiConsumer<Long, Long> onReadEntryFailureCallback) throws InterruptedException, BKException {
+        Map<Integer, BookieId> targetBookieAddresses = null;
+        if (LedgerFragment.ReplicateType.DATA_LOSS == ledgerFragment.getReplicateType()) {
+            Optional<Set<BookieId>> excludedBookies = Optional.empty();
+            targetBookieAddresses = getReplacementBookiesByIndexes(lh, ledgerFragment.getEnsemble(),
+                    ledgerFragment.getBookiesIndexes(), excludedBookies);
+        } else if (LedgerFragment.ReplicateType.DATA_NOT_ADHERING_PLACEMENT == ledgerFragment.getReplicateType()) {
+            targetBookieAddresses = replaceNotAdheringPlacementPolicyBookie(ledgerFragment.getEnsemble(),
+                    lh.getLedgerMetadata().getWriteQuorumSize(), lh.getLedgerMetadata().getAckQuorumSize(),
+                    lh.getLedgerMetadata().getCustomMetadata());
+            ledgerFragment.getBookiesIndexes().addAll(targetBookieAddresses.keySet());
+        }
+        if (MapUtils.isEmpty(targetBookieAddresses)) {
+            LOG.warn("Could not replicate for {} ledger: {}, not find target bookie.",
+                    ledgerFragment.getReplicateType(), ledgerFragment.getLedgerId());
+            return;
+        }
         replicateLedgerFragment(lh, ledgerFragment, targetBookieAddresses, onReadEntryFailureCallback);
     }
 
@@ -1777,6 +1788,13 @@ public PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy(List<BookieI
                 ackQuorumSize);
     }
 
+    public Map<Integer, BookieId> replaceNotAdheringPlacementPolicyBookie(List<BookieId> ensembleBookiesList,
+            int writeQuorumSize, int ackQuorumSize, Map<String, byte[]> customMetadata) {
+        return bkc.getPlacementPolicy()
+                .replaceNotAdheringPlacementPolicyBookie(ensembleBookiesList, writeQuorumSize, ackQuorumSize,
+                        customMetadata);
+    }
+
     /**
      * Makes async request for getting list of entries of ledger from a bookie
      * and returns Future for the result.

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/EnsemblePlacementPolicy.java

@@ -18,6 +18,7 @@
 package org.apache.bookkeeper.client;
 
 import io.netty.util.HashedWheelTimer;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -363,6 +364,20 @@ DistributionSchedule.WriteSet reorderReadLACSequence(
     default void updateBookieInfo(Map<BookieId, BookieInfo> bookieInfoMap) {
     }
 
+    /**
+     * Replace some bookie to adhere placement policy. If the all kinds of replacement
+     * didn't adhere placement policy, return empty map.
+     *
+     * @param ensemble which would be take place.
+     * @param writeQuorumSize
+     * @param ackQuorumSize
+     * @return Map: key means ensemble index, value means target replace bookieId.
+     */
+    default Map<Integer, BookieId> replaceNotAdheringPlacementPolicyBookie(List<BookieId> ensemble, int writeQuorumSize,
+            int ackQuorumSize, Map<String, byte[]> customMetadata) {
+        return Collections.emptyMap();
+    }
+
     /**
      * Select one bookie to the "sticky" bookie where all reads for a particular
      * ledger will be directed to.

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragment.java

@@ -39,8 +39,9 @@ public class LedgerFragment {
     private final long ledgerId;
     private final DistributionSchedule schedule;
     private final boolean isLedgerClosed;
+    private ReplicateType replicateType = ReplicateType.DATA_LOSS;
 
-    LedgerFragment(LedgerHandle lh,
+    public LedgerFragment(LedgerHandle lh,
                    long firstEntryId,
                    long lastKnownEntryId,
                    Set<Integer> bookieIndexes) {
@@ -56,7 +57,7 @@ public class LedgerFragment {
                 || !ensemble.equals(ensembles.get(ensembles.lastKey()));
     }
 
-    LedgerFragment(LedgerFragment lf, Set<Integer> subset) {
+    public LedgerFragment(LedgerFragment lf, Set<Integer> subset) {
         this.ledgerId = lf.ledgerId;
         this.firstEntryId = lf.firstEntryId;
         this.lastKnownEntryId = lf.lastKnownEntryId;
@@ -91,7 +92,7 @@ public boolean isClosed() {
         return isLedgerClosed;
     }
 
-    long getLedgerId() {
+    public long getLedgerId() {
         return ledgerId;
     }
 
@@ -217,11 +218,27 @@ public List<BookieId> getEnsemble() {
         return this.ensemble;
     }
 
+    public ReplicateType getReplicateType() {
+        return replicateType;
+    }
+
+    public void setReplicateType(ReplicateType replicateType) {
+        this.replicateType = replicateType;
+    }
+
     @Override
     public String toString() {
         return String.format("Fragment(LedgerID: %d, FirstEntryID: %d[%d], "
                 + "LastKnownEntryID: %d[%d], Host: %s, Closed: %s)", ledgerId, firstEntryId,
                 getFirstStoredEntryId(), lastKnownEntryId, getLastStoredEntryId(),
                 getAddresses(), isLedgerClosed);
     }
+
+    /**
+     * ReplicateType.
+     */
+    public enum ReplicateType {
+        DATA_LOSS,
+        DATA_NOT_ADHERING_PLACEMENT
+    }
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TopologyAwareEnsemblePlacementPolicy.java

@@ -33,10 +33,12 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo;
 import org.apache.bookkeeper.client.WeightedRandomSelection.WeightedObject;
 import org.apache.bookkeeper.net.BookieId;
@@ -52,6 +54,7 @@
 import org.apache.bookkeeper.stats.Counter;
 import org.apache.bookkeeper.stats.OpStatsLogger;
 import org.apache.bookkeeper.stats.annotations.StatsDoc;
+import org.apache.commons.collections4.CollectionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -788,6 +791,116 @@ public void updateBookieInfo(Map<BookieId, BookieInfo> bookieInfoMap) {
         }
     }
 
+    @Override
+    public Map<Integer, BookieId> replaceNotAdheringPlacementPolicyBookie(List<BookieId> ensemble, int writeQuorumSize,
+            int ackQuorumSize, Map<String, byte[]> customMetadata) {
+        rwLock.readLock().lock();
+        try {
+            if (CollectionUtils.isEmpty(ensemble)) {
+                return Collections.emptyMap();
+            }
+            PlacementPolicyAdherence ensembleAdheringToPlacementPolicy = isEnsembleAdheringToPlacementPolicy(ensemble,
+                    writeQuorumSize, ackQuorumSize);
+            if (PlacementPolicyAdherence.FAIL != ensembleAdheringToPlacementPolicy) {
+                return Collections.emptyMap();
+            }
+            Map<BookieId, Integer> bookieIndex = new HashMap<>();
+            for (int i = 0; i < ensemble.size(); i++) {
+                bookieIndex.put(ensemble.get(i), i);
+            }
+            Map<BookieId, BookieNode> clone = new HashMap<>(knownBookies);
+
+            Map<String, List<BookieNode>> toPlaceGroup = new HashMap<>();
+            for (BookieId bookieId : ensemble) {
+                //When ReplicationWorker.getUnderreplicatedFragments, the bookie is alive, so the fragment is not
+                // data_loss. When find other rack bookie to replace, the bookie maybe shutdown, so here we should pick
+                // the shutdown bookies. If the bookieId shutdown, put it to inactive. When do replace, we should
+                // replace inactive bookie firstly.
+                BookieNode bookieNode = clone.get(bookieId);
+                if (bookieNode == null) {
+                    List<BookieNode> list = toPlaceGroup.computeIfAbsent(NetworkTopology.INACTIVE,
+                            k -> new ArrayList<>());
+                    list.add(new BookieNode(bookieId, NetworkTopology.INACTIVE));
+                } else {
+                    List<BookieNode> list = toPlaceGroup.computeIfAbsent(bookieNode.getNetworkLocation(),
+                            k -> new ArrayList<>());
+                    list.add(bookieNode);
+                }
+            }
+            for (List<BookieNode> bookieNodes : toPlaceGroup.values()) {
+                Collections.shuffle(bookieNodes);
+            }
+
+            Map<String, List<BookieNode>> knownRackToBookies = clone.values().stream()
+                    .collect(Collectors.groupingBy(NodeBase::getNetworkLocation));
+            HashSet<String> knownRacks = new HashSet<>(knownRackToBookies.keySet());
+
+            Set<BookieId> excludesBookies = new HashSet<>();
+
+            for (String key : toPlaceGroup.keySet()) {
+                List<BookieNode> sameRack = knownRackToBookies.get(key);
+                if (!CollectionUtils.isEmpty(sameRack)) {
+                    excludesBookies.addAll(sameRack.stream().map(BookieNode::getAddr).collect(Collectors.toSet()));
+                }
+            }
+
+            Map<Integer, BookieId> targetBookieAddresses = new HashMap<>();
+            boolean placeSucceed = false;
+            while (knownRacks.size() > 0) {
+                BookieNode beReplaceNode = getBeReplaceNode(toPlaceGroup);
+                if (beReplaceNode == null) {
+                    break;
+                }
+                Integer index = bookieIndex.get(beReplaceNode.getAddr());
+                try {
+                    PlacementResult<BookieId> placementResult = replaceBookie(ensemble.size(), writeQuorumSize,
+                            ackQuorumSize, customMetadata, ensemble, beReplaceNode.getAddr(), excludesBookies);
+                    BookieNode replaceNode = clone.get(placementResult.getResult());
+                    String replaceNodeNetwork = replaceNode.getNetworkLocation();
+                    knownRacks.remove(replaceNodeNetwork);
+                    List<BookieNode> nodes = toPlaceGroup.computeIfAbsent(replaceNodeNetwork,
+                            k -> new ArrayList<>());
+                    nodes.add(replaceNode);
+                    targetBookieAddresses.put(index, replaceNode.getAddr());
+                    List<BookieNode> bookieNodes = knownRackToBookies.get(replaceNodeNetwork);
+                    if (!CollectionUtils.isEmpty(bookieNodes)) {
+                        excludesBookies.addAll(
+                                bookieNodes.stream().map(BookieNode::getAddr).collect(Collectors.toSet()));
+                    }
+                } catch (BKException.BKNotEnoughBookiesException e) {
+                    LOG.warn("Didn't find replaced bookie to adhere placement policy.", e);
+                    break;
+                }
+
+                List<BookieId> ensembles = toPlaceGroup.values().stream().flatMap(Collection::stream).map(
+                        BookieNode::getAddr).collect(Collectors.toList());
+                ensembleAdheringToPlacementPolicy = isEnsembleAdheringToPlacementPolicy(ensembles,
+                        writeQuorumSize, ackQuorumSize);
+                if (PlacementPolicyAdherence.FAIL != ensembleAdheringToPlacementPolicy) {
+                    placeSucceed = true;
+                    break;
+                }
+            }
+            return placeSucceed ? targetBookieAddresses : Collections.emptyMap();
+        } finally {
+            rwLock.readLock().unlock();
+        }
+    }
+
+    private BookieNode getBeReplaceNode(Map<String, List<BookieNode>> toPlaceGroup) {
+        List<BookieNode> inactiveNodes = toPlaceGroup.get(NetworkTopology.INACTIVE);
+        if (!CollectionUtils.isEmpty(inactiveNodes)) {
+            return inactiveNodes.remove(inactiveNodes.size() - 1);
+        }
+        Optional<Map.Entry<String, List<BookieNode>>> toPlaceEntry = toPlaceGroup.entrySet().stream()
+                .filter(ele -> ele.getValue().size() > 1).findAny();
+        if (!toPlaceEntry.isPresent()) {
+            return null;
+        }
+        Map.Entry<String, List<BookieNode>> entry = toPlaceEntry.get();
+        return entry.getValue().remove(entry.getValue().size() - 1);
+    }
+
     protected BookieNode createBookieNode(BookieId addr) {
         return new BookieNode(addr, resolveNetworkLocation(addr));
     }
@@ -812,9 +925,9 @@ protected String resolveNetworkLocation(BookieId addr) {
         }
     }
 
-    protected Set<Node> convertBookiesToNodes(Collection<BookieId> excludeBookies) {
+    protected Set<Node> convertBookiesToNodes(Collection<BookieId> bookies) {
         Set<Node> nodes = new HashSet<Node>();
-        for (BookieId addr : excludeBookies) {
+        for (BookieId addr : bookies) {
             BookieNode bn = knownBookies.get(addr);
             if (null == bn) {
                 bn = createBookieNode(addr);

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java

@@ -202,6 +202,8 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
     protected static final String AUDITOR_PERIODIC_BOOKIE_CHECK_INTERVAL = "auditorPeriodicBookieCheckInterval";
     protected static final String AUDITOR_PERIODIC_PLACEMENT_POLICY_CHECK_INTERVAL =
                                                                 "auditorPeriodicPlacementPolicyCheckInterval";
+    protected static final String REPAIRED_PLACEMENT_POLICY_NOT_ADHERING_BOOKIE_ENABLED =
+                                                                "repairedPlacementPolicyNotAdheringBookieEnabled";
     protected static final String AUDITOR_LEDGER_VERIFICATION_PERCENTAGE = "auditorLedgerVerificationPercentage";
     protected static final String AUTO_RECOVERY_DAEMON_ENABLED = "autoRecoveryDaemonEnabled";
     protected static final String LOST_BOOKIE_RECOVERY_DELAY = "lostBookieRecoveryDelay";
@@ -2615,6 +2617,14 @@ public long getAuditorPeriodicPlacementPolicyCheckInterval() {
         return getLong(AUDITOR_PERIODIC_PLACEMENT_POLICY_CHECK_INTERVAL, 0);
     }
 
+    public void setRepairedPlacementPolicyNotAdheringBookieEnable(boolean enabled) {
+        setProperty(REPAIRED_PLACEMENT_POLICY_NOT_ADHERING_BOOKIE_ENABLED, enabled);
+    }
+
+    public boolean getRepairedPlacementPolicyNotAdheringBookieEnable() {
+        return getBoolean(REPAIRED_PLACEMENT_POLICY_NOT_ADHERING_BOOKIE_ENABLED, false);
+    }
+
     /**
      * Sets the grace period (in seconds) for underreplicated ledgers recovery.
      * If ledger is marked underreplicated for more than this period then it

bookkeeper-server/src/main/java/org/apache/bookkeeper/net/NetworkTopology.java

@@ -29,6 +29,7 @@ public interface NetworkTopology {
     String DEFAULT_RACK = "/default-rack";
     String DEFAULT_ZONE = "/default-zone";
     String DEFAULT_UPGRADEDOMAIN = "/default-upgradedomain";
+    String INACTIVE = "/inactive";
     String DEFAULT_ZONE_AND_UPGRADEDOMAIN = DEFAULT_ZONE + DEFAULT_UPGRADEDOMAIN;
     String DEFAULT_REGION_AND_RACK = DEFAULT_REGION + DEFAULT_RACK;
 

bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java

@@ -1470,6 +1470,22 @@ public void process(Long ledgerId, AsyncCallback.VoidCallback iterCallback) {
                             }
                             if (foundSegmentNotAdheringToPlacementPolicy) {
                                 numOfLedgersFoundNotAdheringInPlacementPolicyCheck.incrementAndGet();
+                                //If user enable repaired, mark this ledger to under replication manager.
+                                if (conf.getRepairedPlacementPolicyNotAdheringBookieEnable()) {
+                                    ledgerUnderreplicationManager.markLedgerUnderreplicatedAsync(ledgerId,
+                                            Collections.emptyList()).whenComplete((res, e) -> {
+                                        if (e != null) {
+                                            LOG.error("For ledger: {}, the placement policy not adhering bookie "
+                                                    + "storage, mark it to under replication manager failed.",
+                                                    ledgerId, e);
+                                            return;
+                                        }
+                                        if (LOG.isDebugEnabled()) {
+                                            LOG.debug("For ledger: {}, the placement policy not adhering bookie"
+                                                    + " storage, mark it to under replication manager", ledgerId);
+                                        }
+                                    });
+                                }
                             } else if (foundSegmentSoftlyAdheringToPlacementPolicy) {
                                 numOfLedgersFoundSoftlyAdheringInPlacementPolicyCheck.incrementAndGet();
                             }