improve: remove asserts from code

Signed-off-by: ZhangJian He <[email protected]>

Author: hezhangjian

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

@@ -75,6 +75,9 @@ public EntryKeyValue put(EntryKey k, EntryKeyValue v) {
 
         @Override
         public EntryKeyValue putIfAbsent(EntryKey k, EntryKeyValue v) {
+            if (!k.equals(v)) {
+                throw new IllegalArgumentException("Key and value should be same");
+            }
             assert k.equals(v);
             return super.putIfAbsent(v, v);
         }
@@ -275,11 +278,15 @@ long flushSnapshot(final SkipListFlusher flusher, Checkpoint checkpoint) throws
      */
     void clearSnapshot(final EntrySkipList keyValues) {
         // Caller makes sure that keyValues not empty
-        assert !keyValues.isEmpty();
+        if (keyValues.isEmpty()) {
+            throw new IllegalStateException("Snapshot is empty");
+        }
         this.lock.writeLock().lock();
         try {
+            if (this.snapshot != keyValues) {
+                throw new IllegalStateException("Snapshot mismatch");
+            }
             // create a new snapshot and let the old one go.
-            assert this.snapshot == keyValues;
             this.snapshot = EntrySkipList.EMPTY_VALUE;
         } finally {
             this.lock.writeLock().unlock();
@@ -371,7 +378,9 @@ private EntryKeyValue cloneWithAllocator(long ledgerId, long entryId, final Byte
             return newEntry(ledgerId, entryId, entry);
         }
 
-        assert alloc.getData() != null;
+        if (alloc.getData() == null) {
+            throw new IllegalStateException("MemorySlice.getData() should not be null");
+        }
         entry.get(alloc.getData(), alloc.getOffset(), len);
         return new EntryKeyValue(ledgerId, entryId, alloc.getData(), alloc.getOffset(), len);
     }

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

@@ -571,7 +571,9 @@ void putEntryOffset(long ledger, long entry, long offset) throws IOException {
         LedgerEntryPage lep = null;
         try {
             lep = getLedgerEntryPage(ledger, pageEntry);
-            assert lep != null;
+            if (lep == null) {
+                throw new Bookie.NoLedgerException(ledger);
+            }
             lep.setOffset(offset, offsetInPage * LedgerEntryPage.getIndexEntrySize());
         } catch (FileInfo.FileInfoDeletedException e) {
             throw new Bookie.NoLedgerException(ledger);

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

@@ -45,7 +45,7 @@
  * </p>
  */
 public class SkipListArena {
-    private AtomicReference<Chunk> curChunk = new AtomicReference<Chunk>();
+    private final AtomicReference<Chunk> curChunk = new AtomicReference<Chunk>();
 
     final int chunkSize;
 
@@ -54,7 +54,9 @@ public class SkipListArena {
     public SkipListArena(ServerConfiguration cfg) {
         chunkSize = cfg.getSkipListArenaChunkSize();
         maxAlloc = cfg.getSkipListArenaMaxAllocSize();
-        assert maxAlloc <= chunkSize;
+        if (maxAlloc > chunkSize) {
+            throw new IllegalArgumentException("maxAlloc should be less than or equal to chunkSize");
+        }
     }
 
     /**
@@ -64,7 +66,9 @@ public SkipListArena(ServerConfiguration cfg) {
      * </p>
      */
     public MemorySlice allocateBytes(int size) {
-        assert size >= 0;
+        if (size < 0) {
+            throw new IllegalArgumentException("Invalid size: " + size);
+        }
 
         // Callers should satisfy large allocations directly from JVM since they
         // don't cause fragmentation as badly.
@@ -135,10 +139,10 @@ private static class Chunk {
          * Offset for the next allocation, or the sentinel value -1
          * which implies that the chunk is still uninitialized.
          */
-        private AtomicInteger nextFreeOffset = new AtomicInteger(UNINITIALIZED);
+        private final AtomicInteger nextFreeOffset = new AtomicInteger(UNINITIALIZED);
 
         /** Total number of allocations satisfied from this buffer. */
-        private AtomicInteger allocCount = new AtomicInteger();
+        private final AtomicInteger allocCount = new AtomicInteger();
 
         /** Size of chunk in bytes. */
         private final int size;
@@ -158,17 +162,20 @@ private Chunk(int size) {
          * threads calling alloc(), who will block until the allocation is complete.
          */
         public void init() {
-            assert nextFreeOffset.get() == UNINITIALIZED;
+            if (nextFreeOffset.get() != UNINITIALIZED) {
+                throw new IllegalStateException("Double-init");
+            }
             try {
                 data = new byte[size];
             } catch (OutOfMemoryError e) {
-                boolean failInit = nextFreeOffset.compareAndSet(UNINITIALIZED, OOM);
-                assert failInit; // should be true.
+                nextFreeOffset.compareAndSet(UNINITIALIZED, OOM);
                 throw e;
             }
             // Mark that it's ready for use
             boolean okInit = nextFreeOffset.compareAndSet(UNINITIALIZED, 0);
-            assert okInit;    // single-threaded call
+            if (!okInit) {
+                throw new IllegalStateException("Double-init");
+            }
         }
 
         /**

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

@@ -226,8 +226,6 @@ public void initialBlockingBookieRead() throws BKException {
         CompletableFuture<?> readonly;
         synchronized (this) {
             if (initialReadonlyBookiesFuture == null) {
-                assert initialWritableBookiesFuture == null;
-
                 writable = this.registrationClient.watchWritableBookies(
                             bookies -> processWritableBookiesChanged(bookies.getValue()));
 

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

@@ -158,9 +158,11 @@ private void replicateFragmentInternal(final LedgerHandle lh,
          * INVALID_ENTRY_ID and viceversa.
          */
         if (startEntryId == INVALID_ENTRY_ID ^ endEntryId == INVALID_ENTRY_ID) {
-            LOG.error("For LedgerFragment: {}, seeing inconsistent firstStoredEntryId: {} and lastStoredEntryId: {}",
+            String msg = String.format("For LedgerFragment: %s, seeing inconsistent firstStoredEntryId: %s" +
+                            " and lastStoredEntryId: %s",
                     lf, startEntryId, endEntryId);
-            assert false;
+            LOG.error(msg);
+            throw new IllegalStateException(msg);
         }
 
         if (startEntryId > endEntryId || endEntryId <= INVALID_ENTRY_ID) {
@@ -231,8 +233,7 @@ private void replicateFragmentInternal(final LedgerHandle lh,
     void replicate(final LedgerHandle lh, final LedgerFragment lf,
             final AsyncCallback.VoidCallback ledgerFragmentMcb,
             final Set<BookieId> targetBookieAddresses,
-            final BiConsumer<Long, Long> onReadEntryFailureCallback)
-            throws InterruptedException {
+            final BiConsumer<Long, Long> onReadEntryFailureCallback) {
         Set<LedgerFragment> partitionedFragments = splitIntoSubFragments(lh, lf,
                 bkc.getConf().getRereplicationEntryBatchSize());
         LOG.info("Replicating fragment {} in {} sub fragments.",
@@ -300,9 +301,11 @@ static Set<LedgerFragment> splitIntoSubFragments(LedgerHandle lh,
          * INVALID_ENTRY_ID and viceversa.
          */
         if (firstEntryId == INVALID_ENTRY_ID ^ lastEntryId == INVALID_ENTRY_ID) {
-            LOG.error("For LedgerFragment: {}, seeing inconsistent firstStoredEntryId: {} and lastStoredEntryId: {}",
+            String msg = String.format("For LedgerFragment: %s, seeing inconsistent firstStoredEntryId: %s" +
+                            " and lastStoredEntryId: %s",
                     ledgerFragment, firstEntryId, lastEntryId);
-            assert false;
+            LOG.error(msg);
+            throw new IllegalStateException(msg);
         }
 
         long numberOfEntriesToReplicate = firstEntryId == INVALID_ENTRY_ID ? 0 : (lastEntryId - firstEntryId) + 1;

bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java

@@ -470,11 +470,11 @@ public void remove() {
 
             @Override
             public boolean hasNext() {
-                if (curBatch.size() > 0) {
+                if (!curBatch.isEmpty()) {
                     return true;
                 }
 
-                while (queue.size() > 0 && curBatch.size() == 0) {
+                while (!queue.isEmpty() && curBatch.isEmpty()) {
                     String parent = queue.remove();
                     try {
                         for (String c : zkc.getChildren(parent, false)) {
@@ -501,12 +501,14 @@ public boolean hasNext() {
                         throw new RuntimeException("Error reading list", e);
                     }
                 }
-                return curBatch.size() > 0;
+                return !curBatch.isEmpty();
             }
 
             @Override
             public UnderreplicatedLedger next() {
-                assert curBatch.size() > 0;
+                if (curBatch.isEmpty()) {
+                    throw new IllegalStateException("No more elements");
+                }
                 return curBatch.remove();
             }
         };

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

@@ -89,7 +89,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                 super.channelRead(ctx, msg);
             } else if (msg instanceof BookieProtocol.AuthRequest) { // pre-PB-client
                 BookieProtocol.AuthRequest req = (BookieProtocol.AuthRequest) msg;
-                assert (req.getOpCode() == BookieProtocol.AUTH);
+                if (BookieProtocol.AUTH != req.getOpCode()) {
+                    throw new IllegalStateException("Received message other than auth message");
+                }
                 if (checkAuthPlugin(req.getAuthMessage(), ctx.channel())) {
                     byte[] payload = req
                         .getAuthMessage()
@@ -270,45 +272,44 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-            assert (authProvider != null);
+            if (authProvider == null) {
+                throw new IllegalStateException("Auth provider is not initialized");
+            }
 
             if (authenticated) {
                 super.channelRead(ctx, msg);
             } else if (msg instanceof BookkeeperProtocol.Response) {
                 BookkeeperProtocol.Response resp = (BookkeeperProtocol.Response) msg;
-                if (null == resp.getHeader().getOperation()) {
-                    LOG.info("dropping received malformed message {} from bookie {}", msg, ctx.channel());
-                    // drop the message without header
-                } else {
-                    switch (resp.getHeader().getOperation()) {
-                    case START_TLS:
-                        super.channelRead(ctx, msg);
-                        break;
-                    case AUTH:
-                        if (resp.getStatus() != BookkeeperProtocol.StatusCode.EOK) {
-                            authenticationError(ctx, resp.getStatus().getNumber());
-                        } else {
-                            assert (resp.hasAuthResponse());
-                            BookkeeperProtocol.AuthMessage am = resp.getAuthResponse();
-                            if (AUTHENTICATION_DISABLED_PLUGIN_NAME.equals(am.getAuthPluginName())){
-                                SocketAddress remote = ctx.channel().remoteAddress();
-                                LOG.info("Authentication is not enabled."
+                switch (resp.getHeader().getOperation()) {
+                case START_TLS:
+                    super.channelRead(ctx, msg);
+                    break;
+                case AUTH:
+                    if (resp.getStatus() != BookkeeperProtocol.StatusCode.EOK) {
+                        authenticationError(ctx, resp.getStatus().getNumber());
+                    } else {
+                        if (!resp.hasAuthResponse()) {
+                            throw new IllegalStateException("Auth response is missing in the message");
+                        }
+                        AuthMessage am = resp.getAuthResponse();
+                        if (AUTHENTICATION_DISABLED_PLUGIN_NAME.equals(am.getAuthPluginName())) {
+                            SocketAddress remote = ctx.channel().remoteAddress();
+                            LOG.info("Authentication is not enabled."
                                     + "Considering this client {} authenticated", remote);
-                                AuthHandshakeCompleteCallback cb = new AuthHandshakeCompleteCallback(ctx);
-                                cb.operationComplete(BKException.Code.OK, null);
-                                return;
-                            }
-                            byte[] payload = am.getPayload().toByteArray();
-                            authProvider.process(AuthToken.wrap(payload), new AuthRequestCallback(ctx,
-                                authProviderFactory.getPluginName()));
+                            AuthHandshakeCompleteCallback cb = new AuthHandshakeCompleteCallback(ctx);
+                            cb.operationComplete(BKException.Code.OK, null);
+                            return;
                         }
-                        break;
-                    default:
-                        LOG.warn("dropping received message {} from bookie {}", msg, ctx.channel());
-                        // else just drop the message,
-                        // we're not authenticated so nothing should be coming through
-                        break;
+                        byte[] payload = am.getPayload().toByteArray();
+                        authProvider.process(AuthToken.wrap(payload), new AuthRequestCallback(ctx,
+                                authProviderFactory.getPluginName()));
                     }
+                    break;
+                default:
+                    LOG.warn("dropping received message {} from bookie {}", msg, ctx.channel());
+                    // else just drop the message,
+                    // we're not authenticated so nothing should be coming through
+                    break;
                 }
             } else if (msg instanceof BookieProtocol.Response) {
                 BookieProtocol.Response resp = (BookieProtocol.Response) msg;

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

@@ -237,7 +237,6 @@ boolean hasMasterKey() {
         }
 
         byte[] getMasterKey() {
-            assert hasMasterKey();
             return masterKey;
         }
 

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

@@ -2425,9 +2425,7 @@ abstract class CompletionKey {
         public void release() {}
     }
 
-    /**
-     * Note : Helper functions follow
-     */
+    // Note : Helper functions follow
 
     /**
      * @param status
@@ -2701,7 +2699,9 @@ private void logBookieUnavailable(Runnable logger) {
 
     private void initiateTLS() {
         LOG.info("Initializing TLS to {}", channel);
-        assert state == ConnectionState.CONNECTING;
+        if (ConnectionState.CONNECTING != state) {
+            throw new IllegalStateException("Invalid state to initiate TLS: " + state);
+        }
         final long txnId = getTxnId();
         final CompletionKey completionKey = new TxnCompletionKey(txnId, OperationType.START_TLS);
         completionObjects.put(completionKey,

bookkeeper-server/src/main/java/org/apache/bookkeeper/util/PageCacheUtil.java

@@ -51,16 +51,17 @@ public final class PageCacheUtil {
     }
 
     private static Field getFieldByReflection(Class cls, String fieldName) {
-        Field field = null;
+        Field field;
 
         try {
             field = cls.getDeclaredField(fieldName);
             field.setAccessible(true);
         } catch (Exception e) {
             // We don't really expect this so throw an assertion to
             // catch this during development
-            log.warn("Unable to read {} field from {}", fieldName, cls.getName());
-            assert false;
+            String msg = "Unable to read " + fieldName + " field from " + cls.getName();
+            log.warn(msg);
+            throw new IllegalStateException(msg, e);
         }
 
         return field;

bookkeeper-server/src/main/java/org/apache/bookkeeper/verifier/BookkeeperVerifier.java

@@ -345,7 +345,9 @@ void onLastOpComplete(
         void onLastWriteComplete(
                 Consumer<Integer> cb,
                 Consumer<Consumer<Integer>> newOnLastWrite) {
-            assert (onLastWrite == null);
+            if (onLastWrite != null) {
+                throw new IllegalStateException("onLastWrite already set");
+            }
             onLastWrite = newOnLastWrite;
             checkWriteComplete(cb);
         }

stream/distributedlog/core/src/main/java/org/apache/distributedlog/LocalDLMEmulator.java

@@ -162,15 +162,16 @@ public void run() {
 
     public void start() throws Exception {
         bkStartupThread.start();
-        if (!LocalBookKeeper.waitForServerUp(zkEnsemble, zkTimeoutSec * 1000)) {
+        if (!LocalBookKeeper.waitForServerUp(zkEnsemble, zkTimeoutSec * 1000L)) {
             throw new Exception("Error starting zookeeper/bookkeeper");
         }
         int bookiesUp = checkBookiesUp(numBookies, zkTimeoutSec);
         if (numBookies != bookiesUp) {
-            LOG.info("Only {} bookies are up, expected {} bookies to be there.",
+            String msg = String.format("Only %d bookies are up, expected %d bookies to be there.",
                 bookiesUp, numBookies);
+            LOG.info(msg);
+            throw new IllegalStateException(msg);
         }
-        assert (numBookies == bookiesUp);
         // Provision "/messaging/distributedlog" namespace
         DLMetadata.create(new BKDLConfig(zkEnsemble, "/ledgers")).create(uri);
     }