diff --git a/.github/workflows/backward-compat-tests.yml b/.github/workflows/backward-compat-tests.yml
new file mode 100644
index 00000000000..3789495a957
--- /dev/null
+++ b/.github/workflows/backward-compat-tests.yml
@@ -0,0 +1,58 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+name: Backward compatibility tests
+
+on:
+  push:
+  pull_request:
+    branches:
+      - master
+      - branch-*
+    paths-ignore:
+      - 'site/**'
+    workflow_dispatch:
+
+env:
+  MAVEN_OPTS: -Dmaven.wagon.httpconnectionManager.ttlSeconds=25 -Dmaven.wagon.http.retryHandler.count=3
+
+jobs:
+  test:
+
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+
+      - name: Tune Runner VM
+        uses: ./.github/actions/tune-runner-vm
+
+      - name: Set up JDK 1.8
+        uses: actions/setup-java@v1
+        with:
+          java-version: 1.8
+      - name: Build
+        run: ./gradlew stream:server:build -x test
+      - name: Test current server with old clients
+        run: ./gradlew :tests:backward-compat:current-server-old-clients:test || (tail -n +1 tests/backward-compat/current-server-old-clients/build/reports/tests/test/classes/* && tail -n +1 tests/backward-compat/current-server-old-clients/build/container-logs/**/* && exit 1)
+      - name: Test progressive upgrade
+        run: ./gradlew :tests:backward-compat:upgrade:test || (tail -n +1 tests/backward-compat/upgrade/build/reports/tests/test/classes/* && tail -n +1 tests/backward-compat/upgrade/build/container-logs/**/* && exit 1)
+      - name: Other tests
+        run: ./gradlew :tests:backward-compat:test -x tests:backward-compat:upgrade:test -x :tests:backward-compat:current-server-old-clients:test
diff --git a/bin/common_gradle.sh b/bin/common_gradle.sh
index 3f65c23f3a6..ec704499468 100755
--- a/bin/common_gradle.sh
+++ b/bin/common_gradle.sh
@@ -139,8 +139,8 @@ CLI_GC_OPTS=${CLI_GC_OPTS:-"${DEFAULT_CLI_GC_OPTS}"}
 CLI_GC_LOGGING_OPTS=${CLI_GC_LOGGING_OPTS:-"${DEFAULT_CLI_GC_LOGGING_OPTS}"}
 
 # module names
-BOOKIE_SERVER_MODULE_NAME="bookkeeper-server"
-TABLE_SERVICE_MODULE_NAME="stream-storage-server"
+BOOKIE_SERVER_MODULE_NAME="(org.apache.bookkeeper-)?bookkeeper-server"
+TABLE_SERVICE_MODULE_NAME="(org.apache.bookkeeper-)?stream-storage-server"
 
 is_released_binary() {
   if [ -d ${BK_HOME}/lib ]; then
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java
index 6e31819c2c3..5615338e297 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java
@@ -1035,8 +1035,8 @@ public PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy(List<BookieI
                     LOG.warn("Received exception while trying to get network location of bookie: {}", bookie, e);
                 }
             }
-            if ((racksInQuorum.size() < minNumRacksPerWriteQuorumForThisEnsemble)
-                    || (enforceMinNumRacksPerWriteQuorum && racksInQuorum.contains(getDefaultRack()))) {
+            if (enforceMinNumRacksPerWriteQuorum && (racksInQuorum.size() < minNumRacksPerWriteQuorumForThisEnsemble
+                    || racksInQuorum.contains(getDefaultRack()))) {
                 return PlacementPolicyAdherence.FAIL;
             }
         }
diff --git a/build.gradle b/build.gradle
index 5164f670621..e6147a3e790 100644
--- a/build.gradle
+++ b/build.gradle
@@ -148,6 +148,7 @@ allprojects {
 
         plugins.withType(DistributionPlugin) {
             distTar {
+                archiveClassifier = "bin"
                 compression = Compression.GZIP
                 archiveExtension = 'tar.gz'
             }
@@ -184,7 +185,6 @@ allprojects {
         signing {
             def skipSigning = project.hasProperty('skipSigning') && skipSigning.toBoolean()
             def shouldSign = !skipSigning
-
             if (shouldSign) {
                 if (project.hasProperty("singingKey")) {
                     // The following allow the secretKey and password to be specified using env var
@@ -227,6 +227,12 @@ allprojects {
                     });
                 }
             }
+            systemProperty "gradle.buildDirectory", project.buildDir
+            systemProperty "currentVersion", project.rootProject.version
+            testLogging {
+                outputs.upToDateWhen {false}
+                showStandardStreams = true
+            }
         }
     }
 
diff --git a/settings.gradle b/settings.gradle
index 9cdf810a583..3c72c6e5f80 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -78,6 +78,7 @@ include(':bookkeeper-benchmark',
         'stream:storage:api',
         'stream:storage:impl',
         'stream:tests-common',
+        'tests:docker-images',
         'tests:backward-compat',
         'tests:backward-compat:bc-non-fips',
         'tests:backward-compat:current-server-old-clients',
diff --git a/tests/backward-compat/build.gradle b/tests/backward-compat/build.gradle
index c64d49171e9..7efd6778a43 100644
--- a/tests/backward-compat/build.gradle
+++ b/tests/backward-compat/build.gradle
@@ -38,6 +38,15 @@ subprojects {
         testImplementation depLibs.junit
         testImplementation depLibs.slf4j
         testImplementation depLibs.arquillianCubeDocker
+        testImplementation depLibs.zookeeper
     }
 
+    test {
+        dependsOn(":tests:docker-images:buildImages")
+    }
+
+}
+
+test {
+    subprojects.each { dependsOn("${it.name}:test") }
 }
\ No newline at end of file
diff --git a/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/AbstractClientFencingTest.groovy b/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/AbstractClientFencingTest.groovy
new file mode 100644
index 00000000000..c908d929f4a
--- /dev/null
+++ b/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/AbstractClientFencingTest.groovy
@@ -0,0 +1,97 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.bookkeeper.tests.backwardcompat
+
+import com.github.dockerjava.api.DockerClient
+import org.apache.bookkeeper.tests.integration.utils.BookKeeperClusterUtils
+import org.apache.bookkeeper.tests.integration.utils.MavenClassLoader
+import org.jboss.arquillian.test.api.ArquillianResource
+import org.junit.Assert
+import org.junit.Before
+
+abstract class AbstractClientFencingTest {
+    protected static byte[] PASSWD = "foobar".getBytes()
+
+    @ArquillianResource
+    protected DockerClient docker
+
+    protected String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
+
+    @Before
+    public void before() throws Exception {
+        Assert.assertTrue(BookKeeperClusterUtils.stopAllBookies(docker))
+        // First test to run, formats metadata and bookies
+        if (BookKeeperClusterUtils.metadataFormatIfNeeded(docker, currentVersion)) {
+            BookKeeperClusterUtils.formatAllBookies(docker, currentVersion)
+        }
+        // If already started, this has no effect
+        Assert.assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, currentVersion))
+    }
+
+    protected void testFencingOldClient(String oldClientVersion, String fencingVersion) {
+        String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
+
+        def oldCL = MavenClassLoader.forBookKeeperVersion(oldClientVersion)
+        def oldBK = oldCL.newBookKeeper(zookeeper)
+        def fencingCL = MavenClassLoader.forBookKeeperVersion(fencingVersion)
+        def fencingBK = fencingCL.newBookKeeper(zookeeper)
+
+        try {
+            def numEntries = 5
+            def ledger0 = oldBK.createLedger(3, 2,
+                    oldCL.digestType("CRC32"),
+                    PASSWD)
+            for (int i = 0; i < numEntries; i++) {
+                ledger0.addEntry(("foobar" + i).getBytes())
+            }
+            ledger0.close()
+
+
+            def ledger1 = fencingBK.openLedger(ledger0.getId(), fencingCL.digestType("CRC32"), PASSWD)
+
+            // cannot write any more
+            try {
+                ledger0.addEntry("shouldn't work".getBytes())
+                Assert.fail("Shouldn't have been able to add any more")
+            } catch (Exception e) {
+                Assert.assertEquals(e.getClass().getName(),
+                        "org.apache.bookkeeper.client.BKException\$BKLedgerClosedException")
+            }
+
+            // should be able to open it and read it back
+            def ledger2 = oldBK.openLedger(ledger0.getId(), oldCL.digestType("CRC32"), PASSWD)
+            def entries = ledger2.readEntries(0, ledger2.getLastAddConfirmed())
+            Assert.assertEquals(numEntries, ledger2.getLastAddConfirmed() + 1 /* counts from 0 */)
+            int j = 0
+            while (entries.hasMoreElements()) {
+                def e = entries.nextElement()
+                Assert.assertEquals(new String(e.getEntry()), "foobar"+ j)
+                j++
+            }
+            ledger2.close()
+        } catch(Throwable tt) {
+            throw new RuntimeException("Got exception with oldClientVersion: " + oldClientVersion + " and fencingVersion: " + fencingVersion, tt);
+        } finally {
+            oldBK.close()
+            oldCL.close()
+
+            fencingBK.close()
+            fencingCL.close()
+        }
+    }
+}
diff --git a/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatNewClientFencesOldClient.groovy b/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatNewClientFencesOldClient.groovy
new file mode 100644
index 00000000000..b079891c2b9
--- /dev/null
+++ b/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatNewClientFencesOldClient.groovy
@@ -0,0 +1,37 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.bookkeeper.tests.backwardcompat
+
+import org.apache.bookkeeper.tests.integration.utils.BookKeeperClusterUtils
+import org.apache.bookkeeper.tests.integration.utils.ThreadReaper
+import org.jboss.arquillian.junit.Arquillian
+import org.junit.Test
+import org.junit.runner.RunWith
+
+@RunWith(Arquillian.class)
+class TestCompatNewClientFencesOldClient extends AbstractClientFencingTest {
+
+    @Test
+    public void testNewClientFencesOldClient() throws Exception {
+        BookKeeperClusterUtils.OLD_CLIENT_VERSIONS.each{
+            def version = it
+            ThreadReaper.runWithReaper({ testFencingOldClient(version, currentVersion) })
+        }
+    }
+
+}
diff --git a/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatOldClientFencesOldClient.groovy b/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatOldClientFencesOldClient.groovy
new file mode 100644
index 00000000000..01d7dd81465
--- /dev/null
+++ b/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatOldClientFencesOldClient.groovy
@@ -0,0 +1,40 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.bookkeeper.tests.backwardcompat
+
+import com.github.dockerjava.api.DockerClient
+import org.apache.bookkeeper.tests.integration.utils.BookKeeperClusterUtils
+import org.apache.bookkeeper.tests.integration.utils.ThreadReaper
+import org.jboss.arquillian.junit.Arquillian
+import org.jboss.arquillian.test.api.ArquillianResource
+import org.junit.Assert
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+
+@RunWith(Arquillian.class)
+class TestCompatOldClientFencesOldClient extends AbstractClientFencingTest {
+
+    @Test
+    public void testOldClientFencesOldClient() throws Exception {
+        BookKeeperClusterUtils.OLD_CLIENT_VERSIONS.each{
+            def version = it
+            ThreadReaper.runWithReaper({ testFencingOldClient(version, version) })
+        }
+    }
+}
diff --git a/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatOldClients.groovy b/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatOldClients.groovy
index 9b4abb9abd8..000984a63b2 100644
--- a/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatOldClients.groovy
+++ b/tests/backward-compat/current-server-old-clients/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatOldClients.groovy
@@ -31,25 +31,18 @@ import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
 
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-
 @RunWith(Arquillian.class)
 class TestCompatOldClients {
-    private static final Logger LOG = LoggerFactory.getLogger(TestCompatOldClients.class)
     private static byte[] PASSWD = "foobar".getBytes()
 
-    // 4.1.0 doesn't work because metadata format changed
-    private def oldClientVersions = ["4.4.0", "4.5.1", "4.6.2", "4.7.2", "4.8.2", "4.9.2",
-                                     "4.10.0", "4.11.1", "4.12.1", "4.13.0", "4.14.0" ]
-
     @ArquillianResource
     DockerClient docker
 
-    private String currentVersion = System.getProperty("currentVersion")
+    private String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
 
     @Before
     public void before() throws Exception {
+        Assert.assertTrue(BookKeeperClusterUtils.stopAllBookies(docker))
         // First test to run, formats metadata and bookies
         if (BookKeeperClusterUtils.metadataFormatIfNeeded(docker, currentVersion)) {
             BookKeeperClusterUtils.formatAllBookies(docker, currentVersion)
@@ -58,72 +51,6 @@ class TestCompatOldClients {
         Assert.assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, currentVersion))
     }
 
-    private void testFencingOldClient(String oldClientVersion, String fencingVersion) {
-        String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
-
-        def oldCL = MavenClassLoader.forBookKeeperVersion(oldClientVersion)
-        def oldBK = oldCL.newBookKeeper(zookeeper)
-        def fencingCL = MavenClassLoader.forBookKeeperVersion(fencingVersion)
-        def fencingBK = fencingCL.newBookKeeper(zookeeper)
-
-        try {
-            def numEntries = 5
-            def ledger0 = oldBK.createLedger(3, 2,
-                                             oldCL.digestType("CRC32"),
-                                             PASSWD)
-            for (int i = 0; i < numEntries; i++) {
-                ledger0.addEntry(("foobar" + i).getBytes())
-            }
-            ledger0.close()
-
-
-            def ledger1 = fencingBK.openLedger(ledger0.getId(), fencingCL.digestType("CRC32"), PASSWD)
-
-            // cannot write any more
-            try {
-                ledger0.addEntry("shouldn't work".getBytes())
-                Assert.fail("Shouldn't have been able to add any more")
-            } catch (Exception e) {
-                Assert.assertEquals(e.getClass().getName(),
-                                    "org.apache.bookkeeper.client.BKException\$BKLedgerClosedException")
-            }
-
-            // should be able to open it and read it back
-            def ledger2 = oldBK.openLedger(ledger0.getId(), oldCL.digestType("CRC32"), PASSWD)
-            def entries = ledger2.readEntries(0, ledger2.getLastAddConfirmed())
-            Assert.assertEquals(numEntries, ledger2.getLastAddConfirmed() + 1 /* counts from 0 */)
-            int j = 0
-            while (entries.hasMoreElements()) {
-                def e = entries.nextElement()
-                Assert.assertEquals(new String(e.getEntry()), "foobar"+ j)
-                j++
-            }
-            ledger2.close()
-        } finally {
-            oldBK.close()
-            oldCL.close()
-
-            fencingBK.close()
-            fencingCL.close()
-        }
-    }
-
-    @Test
-    public void testNewClientFencesOldClient() throws Exception {
-        oldClientVersions.each{
-            def version = it
-            ThreadReaper.runWithReaper({ testFencingOldClient(version, currentVersion) })
-        }
-    }
-
-    @Test
-    public void testOldClientFencesOldClient() throws Exception {
-        oldClientVersions.each{
-            def version = it
-            ThreadReaper.runWithReaper({ testFencingOldClient(version, version) })
-        }
-    }
-
     private void testReads(String writeVersion, String readerVersion) throws Exception {
         String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
 
@@ -161,7 +88,7 @@ class TestCompatOldClients {
         }
     }
 
-    private void testReadOpenFailure(String writeVersion, String readerVersion) throws Exception {
+    private void testReadOpenFailure(String writeVersion, String readerVersion, boolean expectFail) throws Exception {
         String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
 
         def writeCL = MavenClassLoader.forBookKeeperVersion(writeVersion)
@@ -180,8 +107,13 @@ class TestCompatOldClients {
 
             try {
                 def ledger1 = readBK.openLedger(ledger0.getId(), readCL.digestType("CRC32"), PASSWD)
-                Assert.fail("For older versions Openledger call is expected to fail with ZKException");
+                if (expectFail) {
+                    Assert.fail("For older versions Openledger call is expected to fail with ZKException, writerVersion: " + writeVersion + ", readerVersion: " + readerVersion);
+                }
             } catch (Exception exc) {
+                if (!expectFail) {
+                    Assert.fail("For older versions Openledger call is expected to work, writerVersion: " + writeVersion + ", readerVersion: " + readerVersion);
+                }
                 Assert.assertEquals(exc.getClass().getName(),
                                 "org.apache.bookkeeper.client.BKException\$ZKException")
             }
@@ -199,15 +131,15 @@ class TestCompatOldClients {
      */
     @Test
     public void testOldClientReadsNewClient() throws Exception {
-        oldClientVersions.each{
+        BookKeeperClusterUtils.OLD_CLIENT_VERSIONS.each{
             def version = it
-            ThreadReaper.runWithReaper({ testReadOpenFailure(currentVersion, version) })
+            ThreadReaper.runWithReaper({ testReadOpenFailure(currentVersion, version, !BookKeeperClusterUtils.hasVersionLatestMetadataFormat(version)) })
         }
     }
 
     @Test
     public void testNewClientReadsNewClient() throws Exception {
-        oldClientVersions.each{
+        BookKeeperClusterUtils.OLD_CLIENT_VERSIONS.each{
             def version = it
             ThreadReaper.runWithReaper({ testReads(version, currentVersion) })
         }
diff --git a/tests/backward-compat/hierarchical-ledger-manager/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatHierarchicalLedgerManager.groovy b/tests/backward-compat/hierarchical-ledger-manager/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatHierarchicalLedgerManager.groovy
index 6394ff0b3f8..69b06d078d2 100644
--- a/tests/backward-compat/hierarchical-ledger-manager/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatHierarchicalLedgerManager.groovy
+++ b/tests/backward-compat/hierarchical-ledger-manager/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatHierarchicalLedgerManager.groovy
@@ -49,7 +49,7 @@ class TestCompatHierarchicalLedgerManager {
      */
     @Test
     public void testCompatHierarchicalLedgerManagerV420toCurrent() throws Exception {
-        String currentVersion = System.getProperty("currentVersion")
+        String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
         BookKeeperClusterUtils.legacyMetadataFormat(docker)
 
         BookKeeperClusterUtils.updateAllBookieConf(docker, "4.2.0",
diff --git a/tests/backward-compat/hostname-bookieid/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeWithHostnameBookieId.groovy b/tests/backward-compat/hostname-bookieid/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeWithHostnameBookieId.groovy
index 4fbdec4eb24..88affbd6919 100644
--- a/tests/backward-compat/hostname-bookieid/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeWithHostnameBookieId.groovy
+++ b/tests/backward-compat/hostname-bookieid/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeWithHostnameBookieId.groovy
@@ -37,9 +37,6 @@ class TestCompatUpgradeWithHostnameBookieId {
     private static final Logger LOG = LoggerFactory.getLogger(TestCompatUpgradeWithHostnameBookieId.class)
     private static byte[] PASSWD = "foobar".getBytes()
 
-    private def oldClientVersions = ["4.4.0", "4.5.1", "4.6.2", "4.7.2", "4.8.2", "4.9.2",
-                                     "4.10.0", "4.11.1", "4.12.1", "4.13.0", "4.14.0" ]
-
     @ArquillianResource
     DockerClient docker
 
@@ -73,7 +70,7 @@ class TestCompatUpgradeWithHostnameBookieId {
         BookKeeperClusterUtils.legacyMetadataFormat(docker)
         String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
 
-        String currentVersion = System.getProperty("currentVersion")
+        String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
 
         Assert.assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, "4.1.0"))
 
@@ -111,7 +108,7 @@ class TestCompatUpgradeWithHostnameBookieId {
             ledger420r.close()
 
             // Ensure we can write and read new ledgers with all client versions
-            oldClientVersions.each{
+            BookKeeperClusterUtils.OLD_CLIENT_VERSIONS.each{
                 LOG.info("Testing ledger creation for version {}", it)
                 def oldCL = MavenClassLoader.forBookKeeperVersion(it)
                 def oldBK = oldCL.newBookKeeper(zookeeper)
diff --git a/tests/backward-compat/old-cookie-new-cluster/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeOldServerInClusterWithCookies.groovy b/tests/backward-compat/old-cookie-new-cluster/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeOldServerInClusterWithCookies.groovy
index dff7d4d56c9..5d9dcb7bb01 100644
--- a/tests/backward-compat/old-cookie-new-cluster/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeOldServerInClusterWithCookies.groovy
+++ b/tests/backward-compat/old-cookie-new-cluster/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeOldServerInClusterWithCookies.groovy
@@ -45,7 +45,7 @@ class TestCompatUpgradeOldServerInClusterWithCookies {
         BookKeeperClusterUtils.legacyMetadataFormat(docker)
         String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
 
-        String currentVersion = System.getProperty("currentVersion")
+        String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
         int numEntries = 10
 
         Assert.assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, "4.1.0"))
diff --git a/tests/backward-compat/recovery-no-password/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatRecoveryNoPassword.groovy b/tests/backward-compat/recovery-no-password/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatRecoveryNoPassword.groovy
index b7987a8356d..dc5f2a3855b 100644
--- a/tests/backward-compat/recovery-no-password/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatRecoveryNoPassword.groovy
+++ b/tests/backward-compat/recovery-no-password/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatRecoveryNoPassword.groovy
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.tests.backwardcompat
 import com.github.dockerjava.api.DockerClient
 
 import io.netty.buffer.ByteBuf
+import org.apache.bookkeeper.net.BookieId
 
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.CountDownLatch
@@ -115,7 +116,7 @@ class TestCompatRecoveryNoPassword {
 
             def cb = new ReplicationVerificationCallback(numRequests)
             for (long i = startEntryId; i < endEntryId; i++) {
-                for (BookieSocketAddress addr : e.getValue()) {
+                for (BookieId addr : e.getValue()) {
                     bookkeeper.getBookieClient()
                         .readEntry(addr, lh.getId(), i, cb, addr, BookieProtocol.FLAG_NONE)
                 }
@@ -143,7 +144,7 @@ class TestCompatRecoveryNoPassword {
         byte[] passwdCorrect = "AAAAAA".getBytes()
         byte[] passwdBad = "BBBBBB".getBytes()
 
-        String currentVersion = System.getProperty("currentVersion")
+        String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
         String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
 
         BookKeeperClusterUtils.legacyMetadataFormat(docker)
@@ -178,7 +179,7 @@ class TestCompatRecoveryNoPassword {
 
             // start a new bookie, and kill one of the initial 2
             def failedBookieId = new BookieSocketAddress(
-                DockerUtils.getContainerIP(docker, bookieContainers.get(0)), 3181)
+                DockerUtils.getContainerIP(docker, bookieContainers.get(0)), 3181).toBookieId()
             Assert.assertTrue(BookKeeperClusterUtils.stopBookie(
                     docker, bookieContainers.get(0)))
             Assert.assertTrue(BookKeeperClusterUtils.startBookieWithVersion(
diff --git a/tests/backward-compat/upgrade-direct/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeDirect.groovy b/tests/backward-compat/upgrade-direct/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeDirect.groovy
index 4a8a95c5ffb..2ee1c7d8aa5 100644
--- a/tests/backward-compat/upgrade-direct/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeDirect.groovy
+++ b/tests/backward-compat/upgrade-direct/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeDirect.groovy
@@ -43,7 +43,7 @@ class TestCompatUpgradeDirect {
     public void test0_upgradeDirect410toCurrent() throws Exception {
         BookKeeperClusterUtils.legacyMetadataFormat(docker)
         String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
-        String currentVersion = System.getProperty("currentVersion")
+        String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
         int numEntries = 10
 
         Assert.assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, "4.1.0"))
@@ -94,7 +94,7 @@ class TestCompatUpgradeDirect {
 
     @Test
     public void test9_v410ClientCantFenceLedgerFromCurrent() throws Exception {
-        String currentVersion = System.getProperty("currentVersion")
+        String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
         String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
 
         def currentCL = MavenClassLoader.forBookKeeperVersion(currentVersion)
diff --git a/tests/backward-compat/upgrade/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgrade.groovy b/tests/backward-compat/upgrade/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgrade.groovy
index 321cd91fc7c..d885e778ff8 100644
--- a/tests/backward-compat/upgrade/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgrade.groovy
+++ b/tests/backward-compat/upgrade/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgrade.groovy
@@ -54,7 +54,7 @@ class TestCompatUpgrade {
         def upgradedBK = upgradedCL.newBookKeeper(zookeeper)
 
         try {
-            def ledger0 = currentRunningBK.createLedger(3, 2,
+            def ledger0 = currentRunningBK.createLedger(2, 2,
                                                         currentRunningCL.digestType("CRC32"),
                                                         PASSWD)
             for (int i = 0; i < numEntries; i++) {
@@ -63,7 +63,7 @@ class TestCompatUpgrade {
             ledger0.close()
 
             // Check whether current client can write to old server
-            def ledger1 = upgradedBK.createLedger(3, 2, upgradedCL.digestType("CRC32"), PASSWD)
+            def ledger1 = upgradedBK.createLedger(2, 2, upgradedCL.digestType("CRC32"), PASSWD)
             try {
                 ledger1.addEntry("foobar".getBytes())
 
@@ -111,161 +111,43 @@ class TestCompatUpgrade {
     }
 
     @Test
-    public void test0() throws Exception {
+    public void test_000() throws Exception {
         BookKeeperClusterUtils.legacyMetadataFormat(docker)
-        Assert.assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, "4.1.0"))
+        Assert.assertTrue(BookKeeperClusterUtils.startAllBookiesWithVersion(docker, "4.8.2"))
     }
 
     @Test
-    public void test410to420() throws Exception {
-        testUpgrade("4.1.0", "4.2.0")
+    public void test_001_482to492() throws Exception {
+        testUpgrade("4.8.2", "4.9.2")
     }
 
     @Test
-    public void test420to421() throws Exception {
-        testUpgrade("4.2.0", "4.2.1")
-    }
-
-    @Test
-    public void test421to422() throws Exception {
-        testUpgrade("4.2.1", "4.2.2")
-    }
-
-    @Test
-    public void test422to423() throws Exception {
-        testUpgrade("4.2.2", "4.2.3")
-    }
-
-    @Test
-    public void test423to424() throws Exception {
-        testUpgrade("4.2.3", "4.2.4")
-    }
-
-    @Test
-    public void test424to430() throws Exception {
-        testUpgrade("4.2.4", "4.3.0", true)
-    }
-
-    @Test
-    public void test430to431() throws Exception {
-        testUpgrade("4.3.0", "4.3.1")
-    }
-
-    @Test
-    public void test431to432() throws Exception {
-        testUpgrade("4.3.1", "4.3.2")
-    }
-
-    @Test
-    public void test432to440() throws Exception {
-        testUpgrade("4.3.2", "4.4.0")
-    }
-
-    @Test
-    public void test440to450() throws Exception {
-        testUpgrade("4.4.0", "4.5.0")
-    }
-
-    @Test
-    public void test450to451() throws Exception {
-        testUpgrade("4.5.0", "4.5.1")
-    }
-
-    @Test
-    public void test451to460() throws Exception {
-        testUpgrade("4.5.1", "4.6.0")
-    }
-
-    @Test
-    public void test460to461() throws Exception {
-        testUpgrade("4.6.0", "4.6.1", false, true)
-    }
-
-    @Test
-    public void test461to462() throws Exception {
-        testUpgrade("4.6.1", "4.6.2", false, true)
-    }
-
-    @Test
-    public void test462to470() throws Exception {
-        testUpgrade("4.6.2", "4.7.0")
-    }
-
-    @Test
-    public void test470to471() throws Exception {
-        testUpgrade("4.7.0", "4.7.1")
-    }
-
-    @Test
-    public void test471to472() throws Exception {
-        testUpgrade("4.7.1", "4.7.2")
-    }
-
-    @Test
-    public void test472to480() throws Exception {
-        testUpgrade("4.7.2", "4.8.0")
-    }
-
-    @Test
-    public void test480to481() throws Exception {
-        testUpgrade("4.8.0", "4.8.1")
-    }
-
-
-    @Test
-    public void test481to482() throws Exception {
-        testUpgrade("4.8.1", "4.8.2")
-    }
-
-    @Test
-    public void test490to491() throws Exception {
-        testUpgrade("4.9.0", "4.9.1")
-    }
-
-
-    @Test
-    public void test491to492() throws Exception {
-        testUpgrade("4.9.1", "4.9.2")
-    }
-
-    @Test
-    public void test492to4100() throws Exception {
+    public void test_002_492to4100() throws Exception {
         testUpgrade("4.9.2", "4.10.0")
     }
 
     @Test
-    public void test4100to4110() throws Exception {
-        testUpgrade("4.10.0", "4.11.0")
-    }
-
-    @Test
-    public void test4110to4111() throws Exception {
-        testUpgrade("4.11.0", "4.11.1")
+    public void test_003_4100to4111() throws Exception {
+        testUpgrade("4.10.0", "4.11.1")
     }
 
     @Test
-    public void test4111to4120() throws Exception {
-        testUpgrade("4.11.1", "4.12.0")
-    }
-
-
-    @Test
-    public void test4120to4121() throws Exception {
-        testUpgrade("4.12.0", "4.12.1")
+    public void test_004_4111to4121() throws Exception {
+        testUpgrade("4.11.1", "4.12.1")
     }
 
     @Test
-    public void test4121to4130() throws Exception {
+    public void test_005_4121to4130() throws Exception {
         testUpgrade("4.12.1", "4.13.0")
     }
 
     @Test
-    public void test4130to4140() throws Exception {
-        testUpgrade("4.13.0", "4.14.0")
+    public void test_006_4130to4143() throws Exception {
+        testUpgrade("4.13.0", "4.14.3")
     }
 
     @Test
-    public void test414toCurrentMaster() throws Exception {
-        testUpgrade("4.14.0", System.getProperty("currentVersion"))
+    public void test_007_4143toCurrentMaster() throws Exception {
+        testUpgrade("4.14.3", BookKeeperClusterUtils.CURRENT_VERSION)
     }
 }
diff --git a/tests/backward-compat/yahoo-custom-version/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeYahooCustom.groovy b/tests/backward-compat/yahoo-custom-version/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeYahooCustom.groovy
index 9762b4bebe1..f1d024a61dc 100644
--- a/tests/backward-compat/yahoo-custom-version/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeYahooCustom.groovy
+++ b/tests/backward-compat/yahoo-custom-version/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatUpgradeYahooCustom.groovy
@@ -125,7 +125,7 @@ class TestCompatUpgradeYahooCustom {
 
     def List<Long> exerciseClients(List<Long> toVerify) {
         String zookeeper = BookKeeperClusterUtils.zookeeperConnectString(docker)
-        String currentVersion = System.getProperty("currentVersion")
+        String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
 
         def ledgers = []
         def yahooCL = MavenClassLoader.forArtifact(yahooRepo, yahooArtifact)
@@ -191,7 +191,7 @@ class TestCompatUpgradeYahooCustom {
 
     @Test
     public void testUpgradeYahooCustom() throws Exception {
-        String currentVersion = System.getProperty("currentVersion")
+        String currentVersion = BookKeeperClusterUtils.CURRENT_VERSION
         String yahooVersion = "4.3-yahoo"
         BookKeeperClusterUtils.metadataFormatIfNeeded(docker, yahooVersion)
 
diff --git a/tests/docker-images/all-released-versions-image/Dockerfile b/tests/docker-images/all-released-versions-image/Dockerfile
index 1cf9c6fd5c1..9c0301fa205 100644
--- a/tests/docker-images/all-released-versions-image/Dockerfile
+++ b/tests/docker-images/all-released-versions-image/Dockerfile
@@ -30,35 +30,15 @@ RUN apt-get update && apt-get install -qy wget supervisor bash \
 
 RUN mkdir /tarballs
 WORKDIR /tarballs
-RUN wget -nv https://archive.apache.org/dist/zookeeper/bookkeeper/bookkeeper-4.0.0/binaries/bookkeeper-server-4.0.0-bin.tar.gz{,.sha1,.md5,.asc}
 RUN wget -nv https://archive.apache.org/dist/zookeeper/bookkeeper/bookkeeper-4.1.0/bookkeeper-server-4.1.0-bin.tar.gz{,.sha1,.md5,.asc}
 RUN wget -nv https://archive.apache.org/dist/zookeeper/bookkeeper/bookkeeper-4.2.0/bookkeeper-server-4.2.0-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/zookeeper/bookkeeper/bookkeeper-4.2.1/bookkeeper-server-4.2.1-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/zookeeper/bookkeeper/bookkeeper-4.2.2/bookkeeper-server-4.2.2-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/zookeeper/bookkeeper/bookkeeper-4.2.3/bookkeeper-server-4.2.3-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/zookeeper/bookkeeper/bookkeeper-4.3.0/bookkeeper-server-4.3.0-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.2.4/bookkeeper-server-4.2.4-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.3.0/bookkeeper-server-4.3.0-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.3.1/bookkeeper-server-4.3.1-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.3.2/bookkeeper-server-4.3.2-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.4.0/bookkeeper-server-4.4.0-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.5.0/bookkeeper-server-4.5.0-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.5.1/bookkeeper-server-4.5.1-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.6.0/bookkeeper-server-4.6.0-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.6.1/bookkeeper-server-4.6.1-bin.tar.gz{,.sha1,.md5,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.6.2/bookkeeper-server-4.6.2-bin.tar.gz{,.sha1,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.7.0/bookkeeper-server-4.7.0-bin.tar.gz{,.sha1,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.7.1/bookkeeper-server-4.7.1-bin.tar.gz{,.sha1,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.7.2/bookkeeper-server-4.7.2-bin.tar.gz{,.sha512,.asc}
 RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.8.2/bookkeeper-server-4.8.2-bin.tar.gz{,.sha512,.asc}
 RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.9.2/bookkeeper-server-4.9.2-bin.tar.gz{,.sha512,.asc}
 RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.10.0/bookkeeper-server-4.10.0-bin.tar.gz{,.sha512,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.11.0/bookkeeper-server-4.11.0-bin.tar.gz{,.sha512,.asc}
 RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.11.1/bookkeeper-server-4.11.1-bin.tar.gz{,.sha512,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.12.0/bookkeeper-server-4.12.0-bin.tar.gz{,.sha512,.asc}
 RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.12.1/bookkeeper-server-4.12.1-bin.tar.gz{,.sha512,.asc}
 RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.13.0/bookkeeper-server-4.13.0-bin.tar.gz{,.sha512,.asc}
-RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.14.0/bookkeeper-server-4.14.0-bin.tar.gz{,.sha512,.asc}
+RUN wget -nv https://archive.apache.org/dist/bookkeeper/bookkeeper-4.14.3/bookkeeper-server-4.14.3-bin.tar.gz{,.sha512,.asc}
 
 RUN wget -nv https://archive.apache.org/dist/incubator/pulsar/pulsar-1.21.0-incubating/apache-pulsar-1.21.0-incubating-bin.tar.gz{,.asc}
 RUN wget -nv https://dist.apache.org/repos/dist/release/bookkeeper/KEYS
diff --git a/tests/docker-images/all-released-versions-image/image_builder.sh b/tests/docker-images/all-released-versions-image/image_builder.sh
new file mode 100755
index 00000000000..a06c29e49b9
--- /dev/null
+++ b/tests/docker-images/all-released-versions-image/image_builder.sh
@@ -0,0 +1,32 @@
+#!/bin/bash
+#
+#/**
+# * Licensed to the Apache Software Foundation (ASF) under one
+# * or more contributor license agreements.  See the NOTICE file
+# * distributed with this work for additional information
+# * regarding copyright ownership.  The ASF licenses this file
+# * to you under the Apache License, Version 2.0 (the
+# * "License"); you may not use this file except in compliance
+# * with the License.  You may obtain a copy of the License at
+# *
+# *     http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# */
+set -e
+IMAGE_NAME=apachebookkeeper/bookkeeper-all-released-versions:latest
+FORCE_REBUILD="${BOOKKEEPER_DOCKER_IMAGES_FORCE_REBUILD:-false}"
+if [[ "$FORCE_REBUILD" != "true" && "$(docker images -q $IMAGE_NAME 2> /dev/null)" != "" ]]; then
+  echo "reusing local image: $IMAGE_NAME"
+  exit 0
+fi
+
+SCRIPT_DIR=$( cd "$(dirname "${BASH_SOURCE[0]}")" ; pwd -P )
+
+## BASE_DIR will be ./bookkeeper/
+BASE_DIR=${SCRIPT_DIR}/../../../
+docker build -t ${IMAGE_NAME} "${BASE_DIR}"/tests/docker-images/all-released-versions-image
\ No newline at end of file
diff --git a/tests/docker-images/all-released-versions-image/scripts/update-conf-and-boot.sh b/tests/docker-images/all-released-versions-image/scripts/update-conf-and-boot.sh
index ba082d616ec..d22d135aa77 100755
--- a/tests/docker-images/all-released-versions-image/scripts/update-conf-and-boot.sh
+++ b/tests/docker-images/all-released-versions-image/scripts/update-conf-and-boot.sh
@@ -29,14 +29,5 @@ sed -i "s|metadataServiceUri=.*|metadataServiceUri=zk+hierarchical://$BK_ZKCONNE
 # 4.7.0 includes journalDirectories instead of `journalDirectory`
 sed -i "s|journalDirectories=.*|journalDirectories=$BK_JOURNALDIR|" /opt/bookkeeper/*/conf/{bk_server,bookkeeper}.conf
 
-# 4.3.1 & 4.3.2 shipped with a broken confs
-sed -i "s|\(# \)\?logSizeLimit=.*|logSizeLimit=1073741824|" /opt/bookkeeper/4.3.1/conf/bk_server.conf
-sed -i "s|\(# \)\?logSizeLimit=.*|logSizeLimit=1073741824|" /opt/bookkeeper/4.3.2/conf/bk_server.conf
-
-# 4.5.1 shipped with a broken conf
-sed -i "s|\(# \)\?statsProviderClass=.*|# disabled stats |" /opt/bookkeeper/4.5.1/conf/bk_server.conf
-
-# 4.6.0 breaks supervisor
-echo "stopasgroup=true" >> /etc/supervisord/conf.d/bookkeeper-4.6.0.conf
 
 exec /usr/bin/supervisord -c /etc/supervisord.conf
diff --git a/tests/docker-images/all-versions-image/image_builder.sh b/tests/docker-images/all-versions-image/image_builder.sh
new file mode 100755
index 00000000000..7b553362303
--- /dev/null
+++ b/tests/docker-images/all-versions-image/image_builder.sh
@@ -0,0 +1,37 @@
+#!/bin/bash
+#
+#/**
+# * Licensed to the Apache Software Foundation (ASF) under one
+# * or more contributor license agreements.  See the NOTICE file
+# * distributed with this work for additional information
+# * regarding copyright ownership.  The ASF licenses this file
+# * to you under the Apache License, Version 2.0 (the
+# * "License"); you may not use this file except in compliance
+# * with the License.  You may obtain a copy of the License at
+# *
+# *     http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# */
+set -e
+IMAGE_NAME=apachebookkeeper/bookkeeper-all-versions:latest
+FORCE_REBUILD="${BOOKKEEPER_DOCKER_IMAGES_FORCE_REBUILD:-false}"
+if [[ "$FORCE_REBUILD" != "true" && "$(docker images -q $IMAGE_NAME 2> /dev/null)" != "" ]]; then
+  echo "reusing local image: $IMAGE_NAME"
+  exit 0
+fi
+
+SCRIPT_DIR=$( cd "$(dirname "${BASH_SOURCE[0]}")" ; pwd -P )
+
+## BASE_DIR will be ./bookkeeper/
+BASE_DIR=${SCRIPT_DIR}/../../../
+VERSION=${1:-UNKNOWN}
+mkdir -p "${BASE_DIR}"/tests/docker-images/all-versions-image/build
+ls -la ${BASE_DIR}bookkeeper-dist/server/build/distributions
+cp ${BASE_DIR}bookkeeper-dist/server/build/distributions/bookkeeper-server-${VERSION}-bin.tar.gz "${BASE_DIR}"/tests/docker-images/all-versions-image/build/
+TARBALL=build/bookkeeper-server-${VERSION}-bin.tar.gz
+docker build -t ${IMAGE_NAME} "${BASE_DIR}"tests/docker-images/all-versions-image --build-arg BK_TARBALL="${TARBALL}"
\ No newline at end of file
diff --git a/tests/docker-images/all-versions-image/scripts/update-conf-and-boot.sh b/tests/docker-images/all-versions-image/scripts/update-conf-and-boot.sh
index 6cddae39561..4248862648b 100644
--- a/tests/docker-images/all-versions-image/scripts/update-conf-and-boot.sh
+++ b/tests/docker-images/all-versions-image/scripts/update-conf-and-boot.sh
@@ -24,14 +24,4 @@ sed -i "s|journalDirectory=.*|journalDirectory=$BK_JOURNALDIR|" /opt/bookkeeper/
 sed -i "s|ledgerDirectories=.*|ledgerDirectories=$BK_LEDGERDIR|" /opt/bookkeeper/*/conf/bk_server.conf
 sed -i "s|zkServers=.*|zkServers=$BK_ZKCONNECTSTRING|" /opt/bookkeeper/*/conf/bk_server.conf
 
-# 4.3.1 & 4.3.2 shipped with a broken confs
-sed -i "s|\(# \)\?logSizeLimit=.*|logSizeLimit=1073741824|" /opt/bookkeeper/4.3.1/conf/bk_server.conf
-sed -i "s|\(# \)\?logSizeLimit=.*|logSizeLimit=1073741824|" /opt/bookkeeper/4.3.2/conf/bk_server.conf
-
-# 4.5.1 shipped with a broken conf
-sed -i "s|\(# \)\?statsProviderClass=.*|# disabled stats |" /opt/bookkeeper/4.5.1/conf/bk_server.conf
-
-# 4.6.0 breaks supervisor
-echo "stopasgroup=true" >> /etc/supervisord/conf.d/bookkeeper-4.6.0.conf
-
 exec /usr/bin/supervisord -c /etc/supervisord.conf
diff --git a/tests/docker-images/build.gradle b/tests/docker-images/build.gradle
new file mode 100644
index 00000000000..20055067148
--- /dev/null
+++ b/tests/docker-images/build.gradle
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+task buildImages(type:Exec) {
+    dependsOn(':stream:server:distTar')
+    dependsOn(':bookkeeper-dist-server:distTar')
+    commandLine './images_builder.sh', project.rootProject.version
+}
diff --git a/tests/docker-images/current-version-image/Dockerfile b/tests/docker-images/current-version-image/Dockerfile
index 4a73f2e9187..9a74fd9b74b 100644
--- a/tests/docker-images/current-version-image/Dockerfile
+++ b/tests/docker-images/current-version-image/Dockerfile
@@ -47,17 +47,17 @@ RUN set -x \
     && yum clean all
 
 # untar tarballs
-ADD target/${DISTRO_NAME}.tar.gz /opt
+ADD build/package/${DISTRO_NAME}.tar.gz /opt
 RUN mv /opt/${PKG_NAME} /opt/bookkeeper
 
 WORKDIR /opt/bookkeeper
 
-COPY target/scripts /opt/bookkeeper/scripts
+COPY build/package/scripts /opt/bookkeeper/scripts
 COPY scripts/install-python-client.sh /opt/bookkeeper/scripts
 RUN chmod +x -R /opt/bookkeeper/scripts/
 
 # copy the python client
-ADD target/bookkeeper-client/ /opt/bookkeeper/bookkeeper-client
+ADD build/package/bookkeeper-client/ /opt/bookkeeper/bookkeeper-client
 RUN /opt/bookkeeper/scripts/install-python-client.sh
 
 ENTRYPOINT [ "/bin/bash", "/opt/bookkeeper/scripts/entrypoint.sh" ]
diff --git a/tests/docker-images/current-version-image/image_builder.sh b/tests/docker-images/current-version-image/image_builder.sh
new file mode 100755
index 00000000000..8719bd5df4a
--- /dev/null
+++ b/tests/docker-images/current-version-image/image_builder.sh
@@ -0,0 +1,46 @@
+#!/bin/bash
+#
+#/**
+# * Licensed to the Apache Software Foundation (ASF) under one
+# * or more contributor license agreements.  See the NOTICE file
+# * distributed with this work for additional information
+# * regarding copyright ownership.  The ASF licenses this file
+# * to you under the Apache License, Version 2.0 (the
+# * "License"); you may not use this file except in compliance
+# * with the License.  You may obtain a copy of the License at
+# *
+# *     http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# */
+set -e
+IMAGE_NAME=apachebookkeeper/bookkeeper-current:latest
+FORCE_REBUILD="${BOOKKEEPER_DOCKER_IMAGES_FORCE_REBUILD:-false}"
+if [[ "$FORCE_REBUILD" != "true" && "$(docker images -q $IMAGE_NAME 2> /dev/null)" != "" ]]; then
+  echo "reusing local image: $IMAGE_NAME"
+  exit 0
+fi
+SCRIPT_DIR=$( cd "$(dirname "${BASH_SOURCE[0]}")" ; pwd -P )
+
+## BASE_DIR will be ./bookkeeper/
+BASE_DIR=${SCRIPT_DIR}/../../../
+
+mkdir -p ${BASE_DIR}/tests/docker-images/current-version-image/build
+OUTPUT_DIR=${BASE_DIR}/tests/docker-images/current-version-image/build/package
+rm -rf $OUTPUT_DIR
+mkdir -p $OUTPUT_DIR
+
+# Python Client
+${BASE_DIR}/stream/clients/python/scripts/docker_build.sh
+
+cp -Rp "${BASE_DIR}stream/clients/python/dist" "${OUTPUT_DIR}/bookkeeper-client"
+cp -Rp "${BASE_DIR}docker/scripts" "${OUTPUT_DIR}/scripts"
+
+VERSION=${1:-UNKNOWN}
+cp -p ${BASE_DIR}bookkeeper-dist/server/build/distributions/bookkeeper-server-${VERSION}-bin.tar.gz "${OUTPUT_DIR}"/bookkeeper-dist-server-${VERSION}-bin.tar.gz
+
+docker build -t ${IMAGE_NAME} --build-arg BK_VERSION="${VERSION}" "${BASE_DIR}"/tests/docker-images/current-version-image
\ No newline at end of file
diff --git a/tests/docker-images/images_builder.sh b/tests/docker-images/images_builder.sh
new file mode 100755
index 00000000000..b000cb2a784
--- /dev/null
+++ b/tests/docker-images/images_builder.sh
@@ -0,0 +1,32 @@
+#!/bin/bash
+#
+#/**
+# * Licensed to the Apache Software Foundation (ASF) under one
+# * or more contributor license agreements.  See the NOTICE file
+# * distributed with this work for additional information
+# * regarding copyright ownership.  The ASF licenses this file
+# * to you under the Apache License, Version 2.0 (the
+# * "License"); you may not use this file except in compliance
+# * with the License.  You may obtain a copy of the License at
+# *
+# *     http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# */
+set -e
+BK_VERSION=$1
+SCRIPT_DIR=$( cd "$(dirname "${BASH_SOURCE[0]}")" ; pwd -P )
+echo "BK_VERSION=${BK_VERSION}"
+
+## BASE_DIR will be ./bookkeeper/
+BASE_DIR=${SCRIPT_DIR}/../../
+cd ${BASE_DIR}
+time /bin/bash -e tests/docker-images/statestore-image/image_builder.sh
+time /bin/bash -e tests/docker-images/all-released-versions-image/image_builder.sh
+time /bin/bash -e tests/docker-images/all-versions-image/image_builder.sh ${BK_VERSION}
+time /bin/bash -e tests/docker-images/current-version-image/image_builder.sh ${BK_VERSION}
+
diff --git a/tests/docker-images/statestore-image/Dockerfile b/tests/docker-images/statestore-image/Dockerfile
index 0c8098a49f6..6d36bdacf6f 100644
--- a/tests/docker-images/statestore-image/Dockerfile
+++ b/tests/docker-images/statestore-image/Dockerfile
@@ -46,8 +46,8 @@ RUN mkdir /opt/bookkeeper/conf
 RUN mkdir /opt/bookkeeper/scripts
 
 ### -----Copy Jars------###
-ADD ./dist/server.tar.gz /opt/
-RUN mv /opt/server/lib/*.jar /opt/bookkeeper/lib/
+ADD ./dist/server-bin.tar.gz /opt/
+RUN mv /opt/server-bin/lib/*.jar /opt/bookkeeper/lib/
 ### --------------------###
 
 ### ----Copy scripts----------###
diff --git a/tests/docker-images/statestore-image/image_builder.sh b/tests/docker-images/statestore-image/image_builder.sh
index 6cc6cde1c46..e17ee608b75 100755
--- a/tests/docker-images/statestore-image/image_builder.sh
+++ b/tests/docker-images/statestore-image/image_builder.sh
@@ -18,22 +18,35 @@
 # * limitations under the License.
 # */
 
+IMAGE_NAME=apachebookkeeper/bookkeeper-current:latest
+FORCE_REBUILD="${BOOKKEEPER_DOCKER_IMAGES_FORCE_REBUILD:-false}"
+if [[ "$FORCE_REBUILD" != "true" && "$(docker images -q $IMAGE_NAME 2> /dev/null)" != "" ]]; then
+  echo "reusing local image: $IMAGE_NAME"
+  exit 0
+fi
 SCRIPT_DIR=$( cd "$(dirname "${BASH_SOURCE[0]}")" ; pwd -P )
 
 ## BASE_DIR will be ./bookkeeper/
 BASE_DIR=${SCRIPT_DIR}/../../../
+rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/dist
+rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/scripts
+rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/temp_conf
+rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/temp_bin
+
 mkdir "${BASE_DIR}"/tests/docker-images/statestore-image/dist
 mkdir "${BASE_DIR}"/tests/docker-images/statestore-image/scripts
 mkdir "${BASE_DIR}"/tests/docker-images/statestore-image/temp_conf
 mkdir "${BASE_DIR}"/tests/docker-images/statestore-image/temp_bin
 
-cp "${BASE_DIR}"/stream/server/build/distributions/server.tar.gz "${BASE_DIR}"/tests/docker-images/statestore-image/dist
+cp "${BASE_DIR}"/stream/server/build/distributions/server-bin.tar.gz "${BASE_DIR}"/tests/docker-images/statestore-image/dist
 cp "${BASE_DIR}"/docker/scripts/* "${BASE_DIR}"/tests/docker-images/statestore-image/scripts
 cp "${BASE_DIR}"/conf/* "${BASE_DIR}"/tests/docker-images/statestore-image/temp_conf
 cp "${BASE_DIR}"/bin/* "${BASE_DIR}"/tests/docker-images/statestore-image/temp_bin
-docker build -t apachebookkeeper/bookkeeper-current:latest "${BASE_DIR}"/tests/docker-images/statestore-image
+docker build -t ${IMAGE_NAME} "${BASE_DIR}"/tests/docker-images/statestore-image
 
 rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/dist
 rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/scripts
 rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/temp_conf
-rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/temp_bin
\ No newline at end of file
+rm -rf "${BASE_DIR}"/tests/docker-images/statestore-image/temp_bin
+
+
diff --git a/tests/integration-tests-topologies/src/main/java/org/apache/bookkeeper/tests/containers/wait/ZKWaitStrategy.java b/tests/integration-tests-topologies/src/main/java/org/apache/bookkeeper/tests/containers/wait/ZKWaitStrategy.java
deleted file mode 100644
index 5b4104110a3..00000000000
--- a/tests/integration-tests-topologies/src/main/java/org/apache/bookkeeper/tests/containers/wait/ZKWaitStrategy.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.bookkeeper.tests.containers.wait;
-
-import java.util.concurrent.TimeUnit;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.bookkeeper.tests.integration.utils.BookKeeperClusterUtils;
-import org.rnorth.ducttape.TimeoutException;
-import org.rnorth.ducttape.unreliables.Unreliables;
-import org.testcontainers.containers.ContainerLaunchException;
-import org.testcontainers.containers.wait.strategy.AbstractWaitStrategy;
-
-/**
- * Wait Strategy until zookeeper container is up.
- */
-@Slf4j
-public class ZKWaitStrategy extends AbstractWaitStrategy {
-
-    private final int zkPort;
-
-    public ZKWaitStrategy(int zkPort) {
-        this.zkPort = zkPort;
-    }
-
-    @Override
-    protected void waitUntilReady() {
-        String hostname = waitStrategyTarget.getContainerIpAddress();
-        int externalPort = waitStrategyTarget.getMappedPort(zkPort);
-
-        try {
-            Unreliables.retryUntilTrue(
-                (int) startupTimeout.getSeconds(),
-                TimeUnit.SECONDS,
-                () -> getRateLimiter().getWhenReady(
-                    () -> {
-                        log.info("Check if zookeeper is running at {}:{}", hostname, externalPort);
-                        return BookKeeperClusterUtils.zookeeperRunning(
-                            hostname, externalPort
-                        );
-                    }));
-        } catch (TimeoutException te) {
-            throw new ContainerLaunchException(
-                "Timed out waiting for zookeeper to be ready");
-        }
-    }
-
-}
diff --git a/tests/integration-tests-topologies/src/main/resources/cube-definitions/3-node-all-version-unstarted.yaml b/tests/integration-tests-topologies/src/main/resources/cube-definitions/3-node-all-version-unstarted.yaml
index c0a2938fc51..a422f4372fc 100644
--- a/tests/integration-tests-topologies/src/main/resources/cube-definitions/3-node-all-version-unstarted.yaml
+++ b/tests/integration-tests-topologies/src/main/resources/cube-definitions/3-node-all-version-unstarted.yaml
@@ -22,7 +22,7 @@ networks:
     driver: bridge
 
 zookeeper*:
-  image: zookeeper:3.4.11
+  image: zookeeper:3.6.2
   await:
     strategy: org.apache.bookkeeper.tests.integration.utils.ZooKeeperAwaitStrategy
   aliases:
@@ -41,6 +41,7 @@ bookkeeper1*:
     - customBeforeStopAction:
         strategy: org.apache.bookkeeper.tests.integration.utils.BookKeeperLogsToTargetDirStopAction
   networkMode: testnetwork*
+  exposedPorts: [3181/tcp, 8080/tcp]
 
 bookkeeper2*:
   image: apachebookkeeper/bookkeeper-all-versions
@@ -51,6 +52,7 @@ bookkeeper2*:
     - customBeforeStopAction:
         strategy: org.apache.bookkeeper.tests.integration.utils.BookKeeperLogsToTargetDirStopAction
   networkMode: testnetwork*
+  exposedPorts: [ 3181/tcp, 8080/tcp ]
 
 bookkeeper3*:
   image: apachebookkeeper/bookkeeper-all-versions
@@ -61,4 +63,5 @@ bookkeeper3*:
     - customBeforeStopAction:
         strategy: org.apache.bookkeeper.tests.integration.utils.BookKeeperLogsToTargetDirStopAction
   networkMode: testnetwork*
+  exposedPorts: [ 3181/tcp, 8080/tcp ]
 
diff --git a/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/BookKeeperClusterUtils.java b/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/BookKeeperClusterUtils.java
index 2d4fc27cc8b..56960061def 100644
--- a/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/BookKeeperClusterUtils.java
+++ b/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/BookKeeperClusterUtils.java
@@ -17,12 +17,9 @@
  */
 package org.apache.bookkeeper.tests.integration.utils;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
-
 import com.github.dockerjava.api.DockerClient;
-
-import java.io.IOException;
-import java.net.Socket;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -31,6 +28,7 @@
 
 import lombok.Cleanup;
 
+import lombok.SneakyThrows;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.apache.zookeeper.ZooDefs.Ids;
@@ -43,8 +41,22 @@
  * Utils for interacting a bookkeeper cluster used for integration tests.
  */
 public class BookKeeperClusterUtils {
+    public static final String CURRENT_VERSION = System.getProperty("currentVersion");
+    public static final List<String> OLD_CLIENT_VERSIONS =
+            Arrays.asList("4.8.2", "4.9.2", "4.10.0", "4.11.1", "4.12.1", "4.13.0", "4.14.3");
+    private static final List<String> OLD_CLIENT_VERSIONS_WITH_CURRENT_LEDGER_METADATA_FORMAT =
+            Arrays.asList("4.9.2", "4.10.0", "4.11.1", "4.12.1", "4.13.0", "4.14.3");
+
+    private static final List<String> OLD_CLIENT_VERSIONS_WITH_OLD_BK_BIN_NAME =
+            Arrays.asList("4.9.2", "4.10.0", "4.11.1", "4.12.1", "4.13.0", "4.14.3", "4.3-yahoo");
+
+
     private static final Logger LOG = LoggerFactory.getLogger(BookKeeperClusterUtils.class);
 
+    public static boolean hasVersionLatestMetadataFormat(String version) {
+        return OLD_CLIENT_VERSIONS_WITH_CURRENT_LEDGER_METADATA_FORMAT.contains(version);
+    }
+
     public static String zookeeperConnectString(DockerClient docker) {
         return DockerUtils.cubeIdsMatching("zookeeper").stream()
             .map((id) -> DockerUtils.getContainerIP(docker, id)).collect(Collectors.joining(":"));
@@ -63,22 +75,19 @@ public static ZooKeeper zookeeperClient(DockerClient docker) throws Exception {
         return zk;
     }
 
+    @SneakyThrows
     public static boolean zookeeperRunning(DockerClient docker, String containerId) {
         String ip = DockerUtils.getContainerIP(docker, containerId);
-        return zookeeperRunning(ip, 2181);
-    }
-    public static boolean zookeeperRunning(String ip, int port) {
-        try (Socket socket = new Socket(ip, port)) {
-            socket.setSoTimeout(1000);
-            socket.getOutputStream().write("ruok".getBytes(UTF_8));
-            byte[] resp = new byte[4];
-            if (socket.getInputStream().read(resp) == 4) {
-                return new String(resp, UTF_8).equals("imok");
-            }
-        } catch (IOException e) {
-            // ignore, we'll return fallthrough to return false
-        }
-        return false;
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        @Cleanup
+        ZooKeeper zk = new ZooKeeper(ip + ":2181", 10000,
+                (e) -> {
+                    if (e.getState().equals(KeeperState.SyncConnected)) {
+                        future.complete(null);
+                    }
+                });
+        future.get();
+        return true;
     }
 
     public static void legacyMetadataFormat(DockerClient docker) throws Exception {
@@ -92,7 +101,7 @@ public static boolean metadataFormatIfNeeded(DockerClient docker, String version
         @Cleanup
         ZooKeeper zk = BookKeeperClusterUtils.zookeeperClient(docker);
         if (zk.exists("/ledgers", false) == null) {
-            String bookkeeper = "/opt/bookkeeper/" + version + "/bin/bookkeeper";
+            String bookkeeper = "/opt/bookkeeper/" + version + "/bin/" + computeBinFilenameByVersion(version);
             runOnAnyBookie(docker, bookkeeper, "shell", "metaformat", "-nonInteractive");
             return true;
         } else {
@@ -122,7 +131,7 @@ public static String createDlogNamespaceIfNeeded(DockerClient docker,
     }
 
     public static void formatAllBookies(DockerClient docker, String version) throws Exception {
-        String bookkeeper = "/opt/bookkeeper/" + version + "/bin/bookkeeper";
+        String bookkeeper = "/opt/bookkeeper/" + version + "/bin/" + computeBinFilenameByVersion(version);
         BookKeeperClusterUtils.runOnAllBookies(docker, bookkeeper, "shell", "bookieformat", "-nonInteractive");
     }
 
@@ -250,4 +259,11 @@ public static boolean waitAllBookieUp(DockerClient docker) {
             .map((b) -> waitBookieUp(docker, b, 10, TimeUnit.SECONDS))
             .reduce(true, BookKeeperClusterUtils::allTrue);
     }
+
+    private static String computeBinFilenameByVersion(String version) {
+        if (OLD_CLIENT_VERSIONS_WITH_OLD_BK_BIN_NAME.contains(version)) {
+            return "bookkeeper";
+        }
+        return "bookkeeper_gradle";
+    }
 }
diff --git a/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/DockerUtils.java b/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/DockerUtils.java
index 415c1b3e373..35d58e42dc9 100644
--- a/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/DockerUtils.java
+++ b/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/DockerUtils.java
@@ -54,9 +54,9 @@ public class DockerUtils {
     private static final Logger LOG = LoggerFactory.getLogger(DockerUtils.class);
 
     private static File getTargetDirectory(String containerId) {
-        String base = System.getProperty("maven.buildDirectory");
+        String base = System.getProperty("gradle.buildDirectory");
         if (base == null) {
-            base = "target";
+            base = "build";
         }
         File directory = new File(base + "/container-logs/" + containerId);
         if (!directory.exists() && !directory.mkdirs()) {
diff --git a/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/MavenClassLoader.java b/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/MavenClassLoader.java
index 2404f21a210..9c24186f920 100644
--- a/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/MavenClassLoader.java
+++ b/tests/integration-tests-utils/src/main/java/org/apache/bookkeeper/tests/integration/utils/MavenClassLoader.java
@@ -21,7 +21,12 @@
 
 import groovy.lang.Closure;
 
+import java.io.Closeable;
 import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
 import java.lang.invoke.MethodHandles;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationHandler;
@@ -29,37 +34,43 @@
 import java.lang.reflect.Proxy;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
+import java.util.zip.GZIPInputStream;
 
+import org.apache.commons.compress.archivers.ArchiveStreamFactory;
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
+import org.apache.commons.io.IOUtils;
 import org.jboss.shrinkwrap.resolver.api.maven.ConfigurableMavenResolverSystem;
 import org.jboss.shrinkwrap.resolver.api.maven.Maven;
 import org.jboss.shrinkwrap.resolver.api.maven.ScopeType;
 import org.jboss.shrinkwrap.resolver.api.maven.coordinate.MavenDependencies;
 import org.jboss.shrinkwrap.resolver.api.maven.coordinate.MavenDependency;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 /**
  * A maven class loader for resolving and loading maven artifacts.
  */
 public class MavenClassLoader implements AutoCloseable {
-    private static final Logger LOG = LoggerFactory.getLogger(MavenClassLoader.class);
 
-    private MavenClassLoader(URLClassLoader cl) {
+    private static List<File> currentVersionLibs;
+
+    private MavenClassLoader(ClassLoader cl) {
         this.classloader = cl;
     }
 
-    private final URLClassLoader classloader;
+    private final ClassLoader classloader;
 
     public static MavenClassLoader forArtifact(String repo, String mainArtifact) throws Exception {
         return createClassLoader(Maven.configureResolver().withRemoteRepo("custom", repo, "default"),
-                                 mainArtifact);
+                mainArtifact);
     }
 
     public static MavenClassLoader forArtifact(String mainArtifact) throws Exception {
@@ -69,55 +80,103 @@ public static MavenClassLoader forArtifact(String mainArtifact) throws Exception
     private static MavenClassLoader createClassLoader(ConfigurableMavenResolverSystem resolver,
                                                       String mainArtifact) throws Exception {
         Optional<String> slf4jVersion = Arrays.stream(resolver.resolve(mainArtifact)
-                                                      .withTransitivity().asResolvedArtifact())
-            .filter((a) -> a.getCoordinate().getGroupId().equals("org.slf4j")
-                    && a.getCoordinate().getArtifactId().equals("slf4j-1.2-api"))
-            .map((a) -> a.getCoordinate().getVersion())
-            .findFirst();
+                        .withTransitivity().asResolvedArtifact())
+                .filter((a) -> a.getCoordinate().getGroupId().equals("org.slf4j")
+                        && a.getCoordinate().getArtifactId().equals("slf4j-1.2-api"))
+                .map((a) -> a.getCoordinate().getVersion())
+                .findFirst();
 
         List<MavenDependency> deps = Lists.newArrayList(
                 MavenDependencies.createDependency(
-                        mainArtifact, ScopeType.COMPILE, false,
-                        MavenDependencies.createExclusion("log4j:log4j")));
+                        mainArtifact, ScopeType.COMPILE, false));
         if (slf4jVersion.isPresent()) {
             deps.add(MavenDependencies.createDependency("org.slf4j:slf4j-simple:" + slf4jVersion.get(),
-                                                        ScopeType.COMPILE, false));
+                    ScopeType.COMPILE, false));
         }
 
         File[] files = resolver.addDependencies(deps.toArray(new MavenDependency[0]))
-            .resolve().withTransitivity().asFile();
+                .resolve().withTransitivity().asFile();
+        return createClassLoader(files);
+    }
+
+    private static MavenClassLoader createClassLoader(File[] jars) {
+        final ClassLoader systemClassLoader = ClassLoader.getSystemClassLoader();
         URLClassLoader cl = AccessController.doPrivileged(
                 new PrivilegedAction<URLClassLoader>() {
                     @Override
                     public URLClassLoader run() {
-                        return new URLClassLoader(Arrays.stream(files)
-                                                  .map((f) -> {
-                                                          try {
-                                                              return f.toURI().toURL();
-                                                          } catch (Throwable t) {
-                                                              throw new RuntimeException(t);
-                                                          }
-                                                      })
-                                                  .toArray(URL[]::new),
-                                                  ClassLoader.getSystemClassLoader());
+                        /**
+                         * Child-first URLClassLoader.
+                         * This is needed because Gradle uses a different version of
+                         * Netty and it is placed in the System Class loader.
+                         */
+                        return new URLClassLoader(Arrays.stream(jars)
+                                .map((f) -> {
+                                    try {
+                                        return f.toURI().toURL();
+                                    } catch (Throwable t) {
+                                        throw new RuntimeException(t);
+                                    }
+                                })
+                                .toArray(URL[]::new),
+                                systemClassLoader) {
+
+                            @Override
+                            protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
+                                Class<?> loadedClass = findLoadedClass(name);
+                                if (loadedClass == null) {
+                                    try {
+                                        loadedClass = findClass(name);
+                                    } catch (ClassNotFoundException ignored) {
+                                    }
+                                    if (loadedClass == null) {
+                                        try {
+                                            loadedClass = systemClassLoader.loadClass(name);
+                                        } catch (ClassNotFoundException e) {
+                                        }
+                                    }
+                                }
+                                if (resolve && loadedClass != null) {
+                                    resolveClass(loadedClass);
+                                }
+                                return loadedClass;
+                            }
+                        };
                     }
                 });
         return new MavenClassLoader(cl);
     }
 
     public static MavenClassLoader forBookKeeperVersion(String version) throws Exception {
-        return forArtifact("org.apache.bookkeeper:bookkeeper-server:" +  version);
+        if (version.equals(BookKeeperClusterUtils.CURRENT_VERSION)) {
+            return forBookkeeperCurrentVersion();
+        }
+        return forArtifact("org.apache.bookkeeper:bookkeeper-server:" + version);
     }
 
-    public Object getClass(String className) throws Exception {
-        return Class.forName(className, true, classloader);
+    private static MavenClassLoader forBookkeeperCurrentVersion() throws Exception {
+        if (currentVersionLibs == null) {
+            final String version = BookKeeperClusterUtils.CURRENT_VERSION;
+            final String artifactName = "bookkeeper-server-" + version + "-bin";
+            final Path tarFile = Paths.get("..", "..", "..",
+                    "bookkeeper-dist", "server", "build", "distributions", artifactName + ".tar.gz");
+            final File tempDir = new File("build");
+            extractTarGz(tarFile.toFile(), tempDir);
+            List<File> jars = new ArrayList<>();
+            Files.list(Paths.get(tempDir.getAbsolutePath(), "bookkeeper-server-" + version, "lib"))
+                    .forEach(path -> {
+                        jars.add(path.toFile());
+                    });
+            currentVersionLibs = jars;
+        }
+        return createClassLoader(currentVersionLibs.toArray(new File[]{}));
     }
 
     public Object callStaticMethod(String className, String methodName, ArrayList<?> args) throws Exception {
         Class<?> klass = Class.forName(className, true, classloader);
 
         try {
-            Class<?>[] paramTypes = args.stream().map((a)-> a.getClass()).toArray(Class[]::new);
+            Class<?>[] paramTypes = args.stream().map((a) -> a.getClass()).toArray(Class[]::new);
             return klass.getMethod(methodName, paramTypes).invoke(null, args.stream().toArray(Object[]::new));
         } catch (NoSuchMethodException nsme) {
             // maybe the params are primitives
@@ -164,12 +223,34 @@ public Object invoke(Object proxy, Method m, Object[] args) throws Throwable {
 
     public Object newInstance(String className, Object... args) throws Exception {
         Class<?> klass = Class.forName(className, true, classloader);
-        return klass.getConstructor(Arrays.stream(args).map((a)-> a.getClass()).toArray(Class[]::new))
-            .newInstance(args);
+        return klass
+                .getConstructor(Arrays.stream(args).map((a) ->
+                                a.getClass())
+                        .toArray(Class[]::new))
+                .newInstance(args);
     }
 
     public Object newBookKeeper(String zookeeper) throws Exception {
-        return newInstance("org.apache.bookkeeper.client.BookKeeper", zookeeper);
+        Class<?> clientConfigurationClass = Class
+                .forName("org.apache.bookkeeper.conf.ClientConfiguration", true, classloader);
+        Object clientConfiguration = newInstance("org.apache.bookkeeper.conf.ClientConfiguration");
+        clientConfigurationClass
+                .getMethod("setZkServers", String.class)
+                .invoke(clientConfiguration, zookeeper);
+
+        // relax timeouts in order to get tests passing in limited environments
+        clientConfigurationClass
+                .getMethod("setReadTimeout", int.class)
+                .invoke(clientConfiguration, 15);
+
+        clientConfigurationClass
+                .getMethod("setZkTimeout", int.class)
+                .invoke(clientConfiguration, 30_000);
+        Class<?> klass = Class.forName("org.apache.bookkeeper.client.BookKeeper", true, classloader);
+        return klass
+                .getConstructor(clientConfigurationClass)
+                .newInstance(clientConfiguration);
+
     }
 
     public Object digestType(String type) throws Exception {
@@ -184,6 +265,54 @@ public Object digestType(String type) throws Exception {
 
     @Override
     public void close() throws Exception {
-        classloader.close();
+        if (classloader instanceof Closeable) {
+            ((Closeable) classloader).close();
+        }
+    }
+
+    private static void extractTarGz(File tarGz, File output) throws Exception {
+        File tarFile = new File(output, tarGz.getName().replace(".gz", ""));
+        tarFile.delete();
+        deCompressGZipFile(tarGz, tarFile);
+        unTar(tarFile, output);
     }
+
+    private static File deCompressGZipFile(File gZippedFile, File tarFile) throws IOException {
+        try (GZIPInputStream gZIPInputStream = new GZIPInputStream(new FileInputStream(gZippedFile));
+             FileOutputStream fos = new FileOutputStream(tarFile)) {
+            byte[] buffer = new byte[1024];
+            int len;
+            while ((len = gZIPInputStream.read(buffer)) > 0) {
+                fos.write(buffer, 0, len);
+            }
+        }
+        return tarFile;
+    }
+
+    private static void unTar(final File inputFile, final File outputDir) throws Exception {
+        try (TarArchiveInputStream debInputStream = (TarArchiveInputStream)
+                new ArchiveStreamFactory().createArchiveInputStream("tar",
+                        new FileInputStream(inputFile))) {
+            TarArchiveEntry entry = null;
+            while ((entry = (TarArchiveEntry) debInputStream.getNextEntry()) != null) {
+                final File outputFile = new File(outputDir, entry.getName());
+                if (entry.isDirectory()) {
+                    if (!outputFile.exists()) {
+                        if (!outputFile.mkdirs()) {
+                            throw new IllegalStateException(
+                                    String.format("Couldn't create directory %s.", outputFile.getAbsolutePath()));
+                        }
+                    } else {
+                        outputFile.delete();
+                    }
+                } else {
+                    try (final OutputStream outputFileStream = new FileOutputStream(outputFile)) {
+                        IOUtils.copy(debInputStream, outputFileStream);
+                    }
+                }
+            }
+        }
+    }
+
+
 }
diff --git a/tests/integration/cluster/build.gradle b/tests/integration/cluster/build.gradle
index f29fedfec84..4cf0dac30d0 100644
--- a/tests/integration/cluster/build.gradle
+++ b/tests/integration/cluster/build.gradle
@@ -45,13 +45,9 @@ dependencies {
     testAnnotationProcessor depLibs.lombok
 }
 
-task buildImage(type:Exec) {
-    workingDir '../../..'
-    commandLine './tests/docker-images/statestore-image/image_builder.sh'
-}
 
 test {
-    dependsOn buildImage
+    dependsOn(":tests:docker-images:buildImages")
     jvmArgs("-Djunit.timeout.test=600000",
             "-Djunit.max.retry=3",
             "-Djava.net.preferIPv4Stack=true",
diff --git a/tests/integration/standalone/build.gradle b/tests/integration/standalone/build.gradle
index 17044f7bce6..fe63dd3b52c 100644
--- a/tests/integration/standalone/build.gradle
+++ b/tests/integration/standalone/build.gradle
@@ -34,13 +34,8 @@ dependencies {
     testAnnotationProcessor depLibs.lombok
 }
 
-task buildImage(type:Exec) {
-    workingDir '../../..'
-    commandLine './tests/docker-images/statestore-image/image_builder.sh'
-}
-
 test {
-    dependsOn buildImage
+    dependsOn(":tests:docker-images:buildImages")
 }
 
 publishing {