Use mockito instead of powermock

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java

@@ -433,7 +433,7 @@ public BookieImpl(ServerConfiguration conf,
         // instantiate the journals
         journals = Lists.newArrayList();
         for (int i = 0; i < journalDirectories.size(); i++) {
-            journals.add(new Journal(i, journalDirectories.get(i),
+            journals.add(Journal.newJournal(i, journalDirectories.get(i),
                     conf, ledgerDirsManager, statsLogger.scope(JOURNAL_SCOPE), allocator, journalAliveListener));
         }
 
@@ -496,6 +496,21 @@ public void ledgerDeleted(long ledgerId) {
         this.bookieStats = new BookieStats(statsLogger, journalDirectories.size(), conf.getJournalQueueSize());
     }
 
+    @VisibleForTesting
+    public static BookieImpl newBookieImpl(ServerConfiguration conf,
+                                           RegistrationManager registrationManager,
+                                           LedgerStorage storage,
+                                           DiskChecker diskChecker,
+                                           LedgerDirsManager ledgerDirsManager,
+                                           LedgerDirsManager indexDirsManager,
+                                           StatsLogger statsLogger,
+                                           ByteBufAllocator allocator,
+                                           Supplier<BookieServiceInfo> bookieServiceInfoProvider)
+            throws IOException, InterruptedException, BookieException {
+        return new BookieImpl(conf, registrationManager, storage, diskChecker,
+                ledgerDirsManager, indexDirsManager, statsLogger, allocator, bookieServiceInfoProvider);
+    }
+
     StateManager initializeStateManager() throws IOException {
         return new BookieStateManager(conf, statsLogger, registrationManager,
                 ledgerDirsManager, bookieServiceInfoProvider);

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java

@@ -19,6 +19,10 @@
 package org.apache.bookkeeper.bookie;
 
 import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithLedgerManagerFactory;
+import static org.apache.bookkeeper.tools.cli.commands.bookie.LastMarkCommand.newLastMarkCommand;
+import static org.apache.bookkeeper.tools.cli.commands.bookies.ClusterInfoCommand.newClusterInfoCommand;
+import static org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand.newListBookiesCommand;
+import static org.apache.bookkeeper.tools.cli.commands.client.SimpleTestCommand.newSimpleTestCommand;
 
 import com.google.common.annotations.VisibleForTesting;
 import java.io.File;
@@ -1007,13 +1011,13 @@ public int runCmd(CommandLine cmdLine) throws Exception {
             int ackQuorum = getOptionIntValue(cmdLine, "ackQuorum", 2);
             int numEntries = getOptionIntValue(cmdLine, "numEntries", 1000);
 
-            SimpleTestCommand.Flags flags = new SimpleTestCommand.Flags()
+            SimpleTestCommand.Flags flags = SimpleTestCommand.Flags.newFlags()
                 .ensembleSize(ensemble)
                 .writeQuorumSize(writeQuorum)
                 .ackQuorumSize(ackQuorum)
                 .numEntries(numEntries);
 
-            SimpleTestCommand command = new SimpleTestCommand(flags);
+            SimpleTestCommand command = newSimpleTestCommand(flags);
 
             command.apply(bkConf, flags);
             return 0;
@@ -1292,7 +1296,7 @@ class LastMarkCmd extends MyCommand {
 
         @Override
         public int runCmd(CommandLine c) throws Exception {
-            LastMarkCommand command = new LastMarkCommand();
+            LastMarkCommand command = newLastMarkCommand();
             command.apply(bkConf, new CliFlags());
             return 0;
         }
@@ -1351,12 +1355,12 @@ public int runCmd(CommandLine cmdLine) throws Exception {
                 return 1;
             }
 
-            ListBookiesCommand.Flags flags = new ListBookiesCommand.Flags()
+            ListBookiesCommand.Flags flags = ListBookiesCommand.Flags.newFlags()
                 .readwrite(readwrite)
                 .readonly(readonly)
                 .all(all);
 
-            ListBookiesCommand command = new ListBookiesCommand(flags);
+            ListBookiesCommand command = newListBookiesCommand(flags);
 
             command.apply(bkConf, flags);
             return 0;
@@ -2552,7 +2556,7 @@ Options getOptions() {
 
         @Override
         int runCmd(CommandLine cmdLine) throws Exception {
-            ClusterInfoCommand cmd = new ClusterInfoCommand();
+            ClusterInfoCommand cmd = newClusterInfoCommand();
             cmd.apply(bkConf, new CliFlags());
             return 0;
         }

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java

@@ -90,7 +90,7 @@ public class GarbageCollectorThread implements Runnable {
 
     // Entry Logger Handle
     final EntryLogger entryLogger;
-    final AbstractLogCompactor compactor;
+    AbstractLogCompactor compactor;
 
     // Stats loggers for garbage collection operations
     private final GarbageCollectorStats gcStats;

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java

@@ -630,18 +630,20 @@ static void writePaddingBytes(JournalChannel jc, ByteBuf paddingBuffer, int jour
 
     // journal entry queue to commit
     final BatchedBlockingQueue<QueueEntry> queue;
-    final BatchedBlockingQueue<ForceWriteRequest> forceWriteRequests;
+    BatchedBlockingQueue<ForceWriteRequest> forceWriteRequests;
 
     volatile boolean running = true;
     private final LedgerDirsManager ledgerDirsManager;
     private final ByteBufAllocator allocator;
-    private final MemoryLimitController memoryLimitController;
 
     // Expose Stats
     private final JournalStats journalStats;
 
     private JournalAliveListener journalAliveListener;
 
+    private MemoryLimitController memoryLimitController;
+
+
     public Journal(int journalIndex, File journalDirectory, ServerConfiguration conf,
             LedgerDirsManager ledgerDirsManager) {
         this(journalIndex, journalDirectory, conf, ledgerDirsManager, NullStatsLogger.INSTANCE,
@@ -724,6 +726,14 @@ public Journal(int journalIndex, File journalDirectory, ServerConfiguration conf
         this.journalAliveListener = journalAliveListener;
     }
 
+    @VisibleForTesting
+    static Journal newJournal(int journalIndex, File journalDirectory, ServerConfiguration conf,
+                                     LedgerDirsManager ledgerDirsManager, StatsLogger statsLogger,
+                                     ByteBufAllocator allocator, JournalAliveListener journalAliveListener) {
+        return new Journal(journalIndex, journalDirectory, conf, ledgerDirsManager, statsLogger, allocator,
+                journalAliveListener);
+    }
+
     JournalStats getJournalStats() {
         return this.journalStats;
     }
@@ -923,6 +933,14 @@ public int getJournalQueueLength() {
         return queue.size();
     }
 
+    @VisibleForTesting
+    JournalChannel newLogFile(long logId, Long replaceLogId) throws IOException {
+        return new JournalChannel(journalDirectory, logId, journalPreAllocSize, journalWriteBufferSize,
+                journalAlignmentSize, removePagesFromCache,
+                journalFormatVersionToWrite, getBufferedChannelBuilder(),
+                conf, fileChannelProvider, replaceLogId);
+    }
+
     /**
      * A thread used for persisting journal entries to journal files.
      *
@@ -991,10 +1009,7 @@ public void run() {
                         ? journalIds.get(0) : null;
 
                     journalCreationWatcher.reset().start();
-                    logFile = new JournalChannel(journalDirectory, logId, journalPreAllocSize, journalWriteBufferSize,
-                                        journalAlignmentSize, removePagesFromCache,
-                                        journalFormatVersionToWrite, getBufferedChannelBuilder(),
-                                        conf, fileChannelProvider, replaceLogId);
+                    logFile = newLogFile(logId, replaceLogId);
 
                     journalStats.getJournalCreationStats().registerSuccessfulEvent(
                             journalCreationWatcher.stop().elapsed(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);
@@ -1275,4 +1290,14 @@ public void joinThread() throws InterruptedException {
     long getMemoryUsage() {
         return memoryLimitController.currentUsage();
     }
+
+    @VisibleForTesting
+    void setMemoryLimitController(MemoryLimitController memoryLimitController) {
+        this.memoryLimitController = memoryLimitController;
+    }
+
+    @VisibleForTesting
+    public void setForceWriteRequests(BatchedBlockingQueue<ForceWriteRequest> forceWriteRequests) {
+        this.forceWriteRequests = forceWriteRequests;
+    }
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LegacyCookieValidation.java

@@ -51,6 +51,11 @@ public LegacyCookieValidation(ServerConfiguration conf,
         this.registrationManager = registrationManager;
     }
 
+    public static LegacyCookieValidation newLegacyCookieValidation(ServerConfiguration conf,
+                                                                   RegistrationManager registrationManager) {
+        return new LegacyCookieValidation(conf, registrationManager);
+    }
+
     @Override
     public void checkCookies(List<File> directories) throws BookieException {
         try {

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/stats/JournalStats.java

@@ -43,6 +43,7 @@
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_SYNC;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_WRITE_BYTES;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.util.function.Supplier;
 import lombok.Getter;
 import org.apache.bookkeeper.bookie.BookKeeperServerStats;
@@ -150,7 +151,7 @@ public class JournalStats {
         name = JOURNAL_NUM_FLUSH_MAX_OUTSTANDING_BYTES,
         help = "The number of journal flushes triggered by MAX_OUTSTANDING_BYTES"
     )
-    private final Counter flushMaxOutstandingBytesCounter;
+    private Counter flushMaxOutstandingBytesCounter;
     @StatsDoc(
         name = JOURNAL_NUM_FLUSH_EMPTY_QUEUE,
         help = "The number of journal flushes triggered when journal queue becomes empty"
@@ -222,4 +223,9 @@ public Long getSample() {
         statsLogger.registerGauge(JOURNAL_MEMORY_USED, journalMemoryUsedStats);
     }
 
+    @VisibleForTesting
+    public void setFlushMaxOutstandingBytesCounter(Counter flushMaxOutstandingBytesCounter) {
+        this.flushMaxOutstandingBytesCounter = flushMaxOutstandingBytesCounter;
+    }
+
 }

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

@@ -24,6 +24,7 @@
 import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithMetadataBookieDriver;
 import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithRegistrationManager;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
@@ -173,6 +174,12 @@ public BookKeeperAdmin(ClientConfiguration conf) throws IOException, Interrupted
         this.mFactory = bkc.ledgerManagerFactory;
     }
 
+    @VisibleForTesting
+    public static BookKeeperAdmin newBookKeeperAdmin(ClientConfiguration conf)
+            throws IOException, InterruptedException, BKException {
+        return new BookKeeperAdmin(conf);
+    }
+
     /**
      * Constructor that takes in a BookKeeper instance . This will be useful,
      * when user already has bk instance ready.

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

@@ -2367,4 +2367,8 @@ void executeOrdered(Runnable runnable) throws RejectedExecutionException {
         executor.execute(runnable);
     }
 
+    @VisibleForTesting
+    public Queue<PendingAddOp> getPendingAddOps() {
+        return pendingAddOps;
+    }
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataBookieDriver.java

@@ -20,6 +20,7 @@
 
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_SCOPE;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -30,6 +31,7 @@
 import org.apache.bookkeeper.meta.exceptions.MetadataException;
 import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.bookkeeper.zookeeper.BoundExponentialBackoffRetryPolicy;
+import org.apache.zookeeper.ZooKeeper;
 
 /**
  * ZooKeeper based metadata bookie driver.
@@ -65,6 +67,12 @@ public synchronized MetadataBookieDriver initialize(ServerConfiguration conf,
 
     @Override
     public synchronized RegistrationManager createRegistrationManager() {
+        ZKRegistrationManager zkRegistrationManager = newZKRegistrationManager(serverConf, zk);
+        return zkRegistrationManager;
+    }
+
+    @VisibleForTesting
+    ZKRegistrationManager newZKRegistrationManager(ServerConfiguration serverConf, ZooKeeper zk) {
         return new ZKRegistrationManager(serverConf, zk);
     }
 

bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java

@@ -18,6 +18,7 @@
  */
 package org.apache.bookkeeper.meta.zk;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import lombok.extern.slf4j.Slf4j;
@@ -31,6 +32,7 @@
 import org.apache.bookkeeper.zookeeper.BoundExponentialBackoffRetryPolicy;
 import org.apache.zookeeper.Watcher.Event.EventType;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooKeeper;
 
 /**
  * ZooKeeper based metadata client driver.
@@ -76,15 +78,21 @@ public synchronized MetadataClientDriver initialize(ClientConfiguration conf,
     @Override
     public synchronized RegistrationClient getRegistrationClient() {
         if (null == regClient) {
-            regClient = new ZKRegistrationClient(
-                zk,
-                ledgersRootPath,
-                scheduler,
-                bookieAddressTracking);
+            regClient = newZKRegistrationClient(
+                    zk,
+                    ledgersRootPath,
+                    scheduler,
+                    bookieAddressTracking);
         }
         return regClient;
     }
 
+    @VisibleForTesting
+    ZKRegistrationClient newZKRegistrationClient(ZooKeeper zk, String ledgersRootPath,
+                                                 ScheduledExecutorService scheduler, boolean bookieAddressTracking) {
+        return new ZKRegistrationClient(zk, ledgersRootPath, scheduler, bookieAddressTracking);
+    }
+
     @Override
     public synchronized void close() {
         if (null != regClient) {

bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java

@@ -108,6 +108,17 @@ public BookieServer(ServerConfiguration conf,
         this.nettyServer.setRequestProcessor(this.requestProcessor);
     }
 
+    @VisibleForTesting
+    public static BookieServer newBookieServer(ServerConfiguration conf,
+                                               Bookie bookie,
+                                               StatsLogger statsLogger,
+                                               ByteBufAllocator allocator,
+                                               UncleanShutdownDetection uncleanShutdownDetection)
+            throws CompatibilityException, UnavailableException, SecurityException, IOException,
+            InterruptedException, KeeperException, BookieException {
+        return new BookieServer(conf, bookie, statsLogger, allocator, uncleanShutdownDetection);
+    }
+
     /**
      * Currently the uncaught exception handler is used for DeathWatcher to notify
      * lifecycle management that a bookie is dead for some reasons.

bookkeeper-server/src/main/java/org/apache/bookkeeper/server/EmbeddedServer.java

@@ -25,6 +25,8 @@
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_SCOPE;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LD_INDEX_SCOPE;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LD_LEDGER_SCOPE;
+import static org.apache.bookkeeper.bookie.BookieImpl.newBookieImpl;
+import static org.apache.bookkeeper.bookie.LegacyCookieValidation.newLegacyCookieValidation;
 import static org.apache.bookkeeper.client.BookKeeperClientStats.CLIENT_SCOPE;
 import static org.apache.bookkeeper.replication.ReplicationStats.REPLICATION_SCOPE;
 import static org.apache.bookkeeper.server.Main.storageDirectoriesFromConf;
@@ -50,7 +52,6 @@
 import org.apache.bookkeeper.bookie.CookieValidation;
 import org.apache.bookkeeper.bookie.LedgerDirsManager;
 import org.apache.bookkeeper.bookie.LedgerStorage;
-import org.apache.bookkeeper.bookie.LegacyCookieValidation;
 import org.apache.bookkeeper.bookie.ReadOnlyBookie;
 import org.apache.bookkeeper.bookie.ScrubberStats;
 import org.apache.bookkeeper.bookie.UncleanShutdownDetection;
@@ -403,8 +404,8 @@ public EmbeddedServer build() throws Exception {
                         registrationManager, integCheck);
                 cookieValidation.checkCookies(storageDirectoriesFromConf(conf.getServerConf()));
             } else {
-                CookieValidation cookieValidation =
-                        new LegacyCookieValidation(conf.getServerConf(), registrationManager);
+                CookieValidation cookieValidation = newLegacyCookieValidation(conf.getServerConf(),
+                        registrationManager);
                 cookieValidation.checkCookies(storageDirectoriesFromConf(conf.getServerConf()));
                 // storage should be created after legacy validation or it will fail (it would find ledger dirs)
                 storage = BookieResources.createLedgerStorage(conf.getServerConf(), ledgerManager,
@@ -419,7 +420,7 @@ public EmbeddedServer build() throws Exception {
                         bookieStats, allocator,
                         bookieServiceInfoProvider);
             } else {
-                bookie = new BookieImpl(conf.getServerConf(), registrationManager, storage,
+                bookie = newBookieImpl(conf.getServerConf(), registrationManager, storage,
                         diskChecker,
                         ledgerDirsManager, indexDirsManager,
                         bookieStats, allocator,

bookkeeper-server/src/main/java/org/apache/bookkeeper/server/service/BookieService.java

@@ -18,6 +18,8 @@
 
 package org.apache.bookkeeper.server.service;
 
+import static org.apache.bookkeeper.proto.BookieServer.newBookieServer;
+
 import java.io.IOException;
 import java.lang.Thread.UncaughtExceptionHandler;
 import java.net.UnknownHostException;
@@ -54,7 +56,7 @@ public BookieService(BookieConfiguration conf,
                          UncleanShutdownDetection uncleanShutdownDetection)
             throws Exception {
         super(NAME, conf, statsLogger);
-        this.server = new BookieServer(conf.getServerConf(),
+        this.server = newBookieServer(conf.getServerConf(),
                                        bookie,
                                        statsLogger,
                                        allocator,