Repository: vitessio/vitess
Pull request #10117 (2022-04-20 13:43:52)

Changes for make tools
Signed-off-by: Rameez Sajwani <rameezwazirali@hotmail.com>

## Description

1. 'Make Tools' on Mac M1 is failing
2. Following unit tests were failing on Mac M1
- go test vitess.io/vitess/go/vt/topo/k8stopo
- go test vitess.io/vitess/go/vt/zkctl
- go test vitess.io/vitess/go/vt/topo/zk2topo
3. Updated versions of zk, etcd, consul for distribution.
- check out my PR https://github.com/vitessio/vitess-resources/pull/8
- New vitess resource release https://github.com/vitessio/vitess-resources/releases/tag/v2.0

## Related Issue(s)
https://github.com/vitessio/vitess/issues/10093

## Checklist

-   [ ] "Backport me!" label has been added if this change should be backported. NO
-   [x] Tests were added or are not required
-   [x] Documentation was added or is not required

## Testing

1. Ran 'Make Tools' on Mac M1 
2. Ran 'Make Tools' on Linux
3. Ran 'Make Tools' on Mac but non-M1 machine.


Patch:
diff --git a/.github/workflows/ensure_bootstrap_updated.yml b/.github/workflows/ensure_bootstrap_updated.yml
index 6dc19c23ed..0cccd47c48 100644
--- a/.github/workflows/ensure_bootstrap_updated.yml
+++ b/.github/workflows/ensure_bootstrap_updated.yml
@@ -19,6 +19,7 @@ jobs:
             - 'docker/**'
             - 'test.go'
             - 'Makefile'
+            - 'bootstrap.sh'
 
     - name: Set up Go
       if: steps.changes.outputs.end_to_end == 'true'
diff --git a/Makefile b/Makefile
index 6d99c5720e..a5bc241de6 100644
--- a/Makefile
+++ b/Makefile
@@ -232,7 +232,6 @@ install_protoc-gen-go:
 PROTO_SRCS = $(wildcard proto/*.proto)
 PROTO_SRC_NAMES = $(basename $(notdir $(PROTO_SRCS)))
 PROTO_GO_OUTS = $(foreach name, $(PROTO_SRC_NAMES), go/vt/proto/$(name)/$(name).pb.go)
-
 # This rule rebuilds all the go files from the proto definitions for gRPC.
 proto: $(PROTO_GO_OUTS)
 
@@ -258,7 +257,7 @@ $(PROTO_GO_OUTS): minimaltools install_protoc-gen-go proto/*.proto
 # This rule builds the bootstrap images for all flavors.
 DOCKER_IMAGES_FOR_TEST = mariadb mariadb103 mysql56 mysql57 mysql80 percona percona57 percona80
 DOCKER_IMAGES = common $(DOCKER_IMAGES_FOR_TEST)
-BOOTSTRAP_VERSION=6
+BOOTSTRAP_VERSION=8
 ensure_bootstrap_version:
 	find docker/ -type f -exec sed -i "s/^\(ARG bootstrap_version\)=.*/\1=${BOOTSTRAP_VERSION}/" {} \;
 	sed -i 's/\(^.*flag.String(\"bootstrap-version\",\) *\"[^\"]\+\"/\1 \"${BOOTSTRAP_VERSION}\"/' test.go
diff --git a/bootstrap.sh b/bootstrap.sh
index c9a7bc22b3..ff978d674a 100755
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -28,7 +28,7 @@ BUILD_CONSUL=${BUILD_CONSUL:-1}
 BUILD_CHROME=${BUILD_CHROME:-1}
 
 VITESS_RESOURCES_DOWNLOAD_BASE_URL="https://github.com/vitessio/vitess-resources/releases/download"
-VITESS_RESOURCES_RELEASE="v1.0"
+VITESS_RESOURCES_RELEASE="v2.0"
 VITESS_RESOURCES_DOWNLOAD_URL="${VITESS_RESOURCES_DOWNLOAD_BASE_URL}/${VITESS_RESOURCES_RELEASE}"
 #
 # 0. Initialization and helper methods.
@@ -56,7 +56,7 @@ install_dep() {
     return
   fi
 
-  echo "installing $name $version"
+  echo "<<< Installing $name $version >>>"
 
   # shellcheck disable=SC2064
   trap "fail '$name build failed'; exit 1" ERR
@@ -100,6 +100,7 @@ install_protoc() {
   case $(uname) in
     Linux)  local platform=linux;;
     Darwin) local platform=osx;;
+    *) echo "ERROR: unsupported platform for protoc"; exit 1;;
   esac
 
   case $(get_arch) in
@@ -107,9 +108,9 @@ install_protoc() {
       x86_64)  local target=x86_64;;
       arm64) case "$platform" in
           osx) local use_homebrew=1;;
-          *) echo "ERROR: unsupported architecture"; exit 1;;
+          *) echo "ERROR: unsupported architecture for protoc"; exit 1;;
       esac;;
-      *)   echo "ERROR: unsupported architecture"; exit 1;;
+      *)   echo "ERROR: unsupported architecture for protoc"; exit 1;;
   esac
 
   # TODO (ajm188): remove this branch after protoc includes signed protoc binaries for M1 Macs.
@@ -151,18 +152,17 @@ WARNING
 install_zookeeper() {
   local version="$1"
   local dist="$2"
-
   zk="zookeeper-$version"
+  vtzk="vt-zookeeper-$version"
   # This is how we'd download directly from source:
-  # wget "https://archive.apache.org/dist/zookeeper/$zk/$zk.tar.gz"
-  $VTROOT/tools/wget-retry "${VITESS_RESOURCES_DOWNLOAD_URL}/${zk}.tar.gz"
-  tar -xzf "$zk.tar.gz"
-  ant -f "$zk/build.xml" package
-  ant -f "$zk/zookeeper-contrib/zookeeper-contrib-fatjar/build.xml" jar
-  mkdir -p lib
-  cp "$zk/build/zookeeper-contrib/zookeeper-contrib-fatjar/zookeeper-dev-fatjar.jar" "lib/$zk-fatjar.jar"
-  zip -d "lib/$zk-fatjar.jar" 'META-INF/*.SF' 'META-INF/*.RSA' 'META-INF/*SF' || true # needed for >=3.4.10 <3.5
-  rm -rf "$zk" "$zk.tar.gz"
+  # wget "https://dlcdn.apache.org/zookeeper/$zk/apache-$zk.tar.gz"
+  $VTROOT/tools/wget-retry "${VITESS_RESOURCES_DOWNLOAD_URL}/apache-${zk}.tar.gz"
+  tar -xzf "$dist/apache-$zk.tar.gz"
+  mv $dist/apache-$zk $dist/$vtzk
+  mvn -f $dist/$vtzk/zookeeper-contrib/zookeeper-contrib-fatjar/pom.xml clean install -P fatjar -DskipTests
+  mkdir -p $dist/$vtzk/lib
+  cp "$dist/$vtzk/zookeeper-contrib/zookeeper-contrib-fatjar/target/$zk-fatjar.jar" "$dist/$vtzk/lib/$zk-fatjar.jar"
+  rm -rf "$zk.tar.gz"
 }
 
 
@@ -174,12 +174,14 @@ install_etcd() {
   case $(uname) in
     Linux)  local platform=linux; local ext=tar.gz;;
     Darwin) local platform=darwin; local ext=zip;;
+    *)   echo "ERROR: unsupported platform for etcd"; exit 1;;
   esac
 
   case $(get_arch) in
       aarch64)  local target=arm64;;
       x86_64)  local target=amd64;;
-      *)   echo "ERROR: unsupported architecture"; exit 1;;
+      arm64)  local target=arm64;;
+      *)   echo "ERROR: unsupported architecture for etcd"; exit 1;;
   esac
 
   file="etcd-${version}-${platform}-${target}.${ext}"
@@ -203,7 +205,6 @@ install_etcd() {
 install_k3s() {
   local version="$1"
   local dist="$2"
-
   case $(uname) in
     Linux)  local platform=linux;;
     *)   echo "WARNING: unsupported platform. K3s only supports running on Linux, the k8s topology will not be available for local examples."; return;;
@@ -211,7 +212,8 @@ install_k3s() {
 
   case $(get_arch) in
       aarch64)  local target="-arm64";;
-      x86_64)  local target="";;
+      x86_64) local target="";;
+      arm64)  local target="-arm64";;
       *)   echo "WARNING: unsupported architecture, the k8s topology will not be available for local examples."; return;;
   esac
 
@@ -235,12 +237,14 @@ install_consul() {
   case $(uname) in
     Linux)  local platform=linux;;
     Darwin) local platform=darwin;;
+    *)   echo "ERROR: unsupported platform for consul"; exit 1;;
   esac
 
   case $(get_arch) in
-      aarch64)  local target=arm64;;
-      x86_64)  local target=amd64;;
-      *)   echo "ERROR: unsupported architecture"; exit 1;;
+    aarch64)  local target=arm64;;
+    x86_64)  local target=amd64;;
+    arm64)  local target=arm64;;
+    *)   echo "ERROR: unsupported architecture for consul"; exit 1;;
   esac
 
   # This is how we'd download directly from source:
@@ -284,25 +288,27 @@ install_chromedriver() {
 }
 
 install_all() {
+  echo "##local system details..."
+  echo "##platform: $(uname) target:$(get_arch) OS: $os"
   # protoc
   protoc_ver=3.19.4
   install_dep "protoc" "$protoc_ver" "$VTROOT/dist/vt-protoc-$protoc_ver" install_protoc
 
   # zk
-  zk_ver=${ZK_VERSION:-3.4.14}
+  zk_ver=${ZK_VERSION:-3.8.0}
   if [ "$BUILD_JAVA" == 1 ] ; then
-    install_dep "Zookeeper" "$zk_ver" "$VTROOT/dist/vt-zookeeper-$zk_ver" install_zookeeper
+    install_dep "Zookeeper" "$zk_ver" "$VTROOT/dist" install_zookeeper
   fi
 
   # etcd
-  command -v etcd && echo "etcd already installed" || install_dep "etcd" "v3.3.10" "$VTROOT/dist/etcd" install_etcd
+  command -v etcd && echo "etcd already installed" || install_dep "etcd" "v3.5.3" "$VTROOT/dist/etcd" install_etcd
 
   # k3s
   command -v  k3s || install_dep "k3s" "v1.0.0" "$VTROOT/dist/k3s" install_k3s
 
   # consul
   if [ "$BUILD_CONSUL" == 1 ] ; then
-    install_dep "Consul" "1.4.0" "$VTROOT/dist/consul" install_consul
+    install_dep "Consul" "1.11.4" "$VTROOT/dist/consul" install_consul
   fi
 
   # chromedriver
diff --git a/config/zkcfg/zoo.cfg b/config/zkcfg/zoo.cfg
index 8b1b009941..b574a0f897 100644
--- a/config/zkcfg/zoo.cfg
+++ b/config/zkcfg/zoo.cfg
@@ -4,6 +4,8 @@ clientPort={{.ClientPort}}
 initLimit=5
 syncLimit=2
 maxClientCnxns=0
+# enable commands like ruok by default
+4lw.commands.whitelist=*
 {{range .Servers}}
 server.{{.ServerId}}={{.Hostname}}:{{.LeaderPort}}:{{.ElectionPort}}
 {{end}}
diff --git a/docker/base/Dockerfile b/docker/base/Dockerfile
index 6b859ea7c4..8b7cf00891 100644
--- a/docker/base/Dockerfile
+++ b/docker/base/Dockerfile
@@ -21,7 +21,7 @@
 # TODO(mberlin): Remove the symlink and this note once
 # https://github.com/docker/hub-feedback/issues/292 is fixed.
 
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql57"
 
 FROM "${image}"
diff --git a/docker/base/Dockerfile.mariadb b/docker/base/Dockerfile.mariadb
index bc09751498..7317acfa03 100644
--- a/docker/base/Dockerfile.mariadb
+++ b/docker/base/Dockerfile.mariadb
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mariadb"
 
 FROM "${image}"
diff --git a/docker/base/Dockerfile.mariadb103 b/docker/base/Dockerfile.mariadb103
index 2053299ac7..7ed4ce28c0 100644
--- a/docker/base/Dockerfile.mariadb103
+++ b/docker/base/Dockerfile.mariadb103
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mariadb103"
 
 FROM "${image}"
diff --git a/docker/base/Dockerfile.mysql56 b/docker/base/Dockerfile.mysql56
index 028b0d55d8..4b5d752be7 100644
--- a/docker/base/Dockerfile.mysql56
+++ b/docker/base/Dockerfile.mysql56
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql56"
 
 FROM "${image}"
diff --git a/docker/base/Dockerfile.mysql80 b/docker/base/Dockerfile.mysql80
index 32fec95057..5aebc41d3f 100644
--- a/docker/base/Dockerfile.mysql80
+++ b/docker/base/Dockerfile.mysql80
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql80"
 
 FROM "${image}"
diff --git a/docker/base/Dockerfile.percona b/docker/base/Dockerfile.percona
index 90f7ea8e68..1b9be4c3cd 100644
--- a/docker/base/Dockerfile.percona
+++ b/docker/base/Dockerfile.percona
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-percona"
 
 FROM "${image}"
diff --git a/docker/base/Dockerfile.percona57 b/docker/base/Dockerfile.percona57
index 655af7e112..fb0b914cc7 100644
--- a/docker/base/Dockerfile.percona57
+++ b/docker/base/Dockerfile.percona57
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-percona57"
 
 FROM "${image}"
diff --git a/docker/base/Dockerfile.percona80 b/docker/base/Dockerfile.percona80
index d75b4339f4..7a49514a54 100644
--- a/docker/base/Dockerfile.percona80
+++ b/docker/base/Dockerfile.percona80
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-percona80"
 
 FROM "${image}"
diff --git a/docker/lite/Dockerfile.alpine b/docker/lite/Dockerfile.alpine
index 7bba5c3516..8298f8c119 100644
--- a/docker/lite/Dockerfile.alpine
+++ b/docker/lite/Dockerfile.alpine
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mariadb103"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.mariadb b/docker/lite/Dockerfile.mariadb
index 47ab283a22..58e94bb31f 100644
--- a/docker/lite/Dockerfile.mariadb
+++ b/docker/lite/Dockerfile.mariadb
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mariadb"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.mariadb103 b/docker/lite/Dockerfile.mariadb103
index 67fed10e2b..1a0ef5c0c5 100644
--- a/docker/lite/Dockerfile.mariadb103
+++ b/docker/lite/Dockerfile.mariadb103
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mariadb103"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.mysql56 b/docker/lite/Dockerfile.mysql56
index 4b41359bd3..e3192acc18 100644
--- a/docker/lite/Dockerfile.mysql56
+++ b/docker/lite/Dockerfile.mysql56
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql56"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.mysql57 b/docker/lite/Dockerfile.mysql57
index c05f72a38e..eb09b7b614 100644
--- a/docker/lite/Dockerfile.mysql57
+++ b/docker/lite/Dockerfile.mysql57
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql57"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.mysql80 b/docker/lite/Dockerfile.mysql80
index 14a2b980fc..c8a958fdbe 100644
--- a/docker/lite/Dockerfile.mysql80
+++ b/docker/lite/Dockerfile.mysql80
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql80"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.percona b/docker/lite/Dockerfile.percona
index 5769d6cca4..32afc6f60c 100644
--- a/docker/lite/Dockerfile.percona
+++ b/docker/lite/Dockerfile.percona
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-percona"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.percona57 b/docker/lite/Dockerfile.percona57
index ef4da25374..20c6eaf67f 100644
--- a/docker/lite/Dockerfile.percona57
+++ b/docker/lite/Dockerfile.percona57
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-percona57"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.percona80 b/docker/lite/Dockerfile.percona80
index dce2dcc2a0..d71f149729 100644
--- a/docker/lite/Dockerfile.percona80
+++ b/docker/lite/Dockerfile.percona80
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-percona80"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.ubi7.mysql57 b/docker/lite/Dockerfile.ubi7.mysql57
index 2c2af252f9..f14813f9a3 100644
--- a/docker/lite/Dockerfile.ubi7.mysql57
+++ b/docker/lite/Dockerfile.ubi7.mysql57
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql57"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.ubi7.mysql80 b/docker/lite/Dockerfile.ubi7.mysql80
index 1fab47f1f2..6fea92a584 100644
--- a/docker/lite/Dockerfile.ubi7.mysql80
+++ b/docker/lite/Dockerfile.ubi7.mysql80
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql80"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.ubi7.percona57 b/docker/lite/Dockerfile.ubi7.percona57
index d5bc02e485..539f9b6963 100644
--- a/docker/lite/Dockerfile.ubi7.percona57
+++ b/docker/lite/Dockerfile.ubi7.percona57
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-percona57"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.ubi7.percona80 b/docker/lite/Dockerfile.ubi7.percona80
index 1ea48e48c2..31c827d382 100644
--- a/docker/lite/Dockerfile.ubi7.percona80
+++ b/docker/lite/Dockerfile.ubi7.percona80
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-percona80"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.ubi8.arm64.mysql80 b/docker/lite/Dockerfile.ubi8.arm64.mysql80
index 591ce72a69..3bd1af4bcb 100644
--- a/docker/lite/Dockerfile.ubi8.arm64.mysql80
+++ b/docker/lite/Dockerfile.ubi8.arm64.mysql80
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql80"
 
 FROM "${image}" AS builder
diff --git a/docker/lite/Dockerfile.ubi8.mysql80 b/docker/lite/Dockerfile.ubi8.mysql80
index 4895c256d3..1a93749fb1 100644
--- a/docker/lite/Dockerfile.ubi8.mysql80
+++ b/docker/lite/Dockerfile.ubi8.mysql80
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql80"
 
 FROM "${image}" AS builder
diff --git a/docker/local/Dockerfile b/docker/local/Dockerfile
index b2b833a46e..923ded12ec 100644
--- a/docker/local/Dockerfile
+++ b/docker/local/Dockerfile
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-common"
 
 FROM "${image}"
diff --git a/go/vt/orchestrator/logic/orchestrator.go b/go/vt/orchestrator/logic/orchestrator.go
index d181e223bf..1d9b4e6c74 100644
--- a/go/vt/orchestrator/logic/orchestrator.go
+++ b/go/vt/orchestrator/logic/orchestrator.go
@@ -51,8 +51,6 @@ var discoveryQueue *discovery.Queue
 var snapshotDiscoveryKeys chan inst.InstanceKey
 var snapshotDiscoveryKeysMutex sync.Mutex
 var hasReceivedSIGTERM int32
-var ersInProgressMutex sync.Mutex
-var ersInProgress bool
 
 var discoveriesCounter = metrics.NewCounter()
 var failedDiscoveriesCounter = metrics.NewCounter()
diff --git a/go/vt/orchestrator/logic/topology_recovery.go b/go/vt/orchestrator/logic/topology_recovery.go
index 223852458e..d5b77fbda1 100644
--- a/go/vt/orchestrator/logic/topology_recovery.go
+++ b/go/vt/orchestrator/logic/topology_recovery.go
@@ -19,7 +19,6 @@ package logic
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"math/rand"
 	goos "os"
@@ -619,10 +618,44 @@ func checkAndRecoverDeadPrimary(analysisEntry inst.ReplicationAnalysis, candidat
 	if !(forceInstanceRecovery || analysisEntry.ClusterDetails.HasAutomatedPrimaryRecovery) {
 		return false, nil, nil
 	}
-	tablet, err := TabletRefresh(analysisEntry.AnalyzedInstanceKey)
+
+	// We lock the shard here and then refresh the tablets information
+	ctx, unlock, err := LockShard(context.Background(), analysisEntry.AnalyzedInstanceKey)
 	if err != nil {
 		return false, nil, err
 	}
+	defer unlock(&err)
+
+	// TODO (@GuptaManan100): Refresh only the shard tablet information instead of all the tablets
+	RefreshTablets(true /* forceRefresh */)
+
+	// Run a replication analysis again. We need this because vtorc works on ephemeral data to find the failure scenarios.
+	// That data might be old, because of a cluster operation that was run through vtctld or some other vtorc. So before we do any
+	// changes, we should be checking that this failure is indeed needed to be fixed. We do this after locking the shard to be sure
+	// that the data that we use now is up-to-date.
+	analysisEntries, err := inst.GetReplicationAnalysis(analysisEntry.ClusterDetails.ClusterName, &inst.ReplicationAnalysisHints{})
+	if err != nil {
+		return false, nil, err
+	}
+
+	// The recovery is only required if the same instance key requires a DeadPrimary or DeadPrimaryAndSomeReplicas recovery.
+	recoveryRequired := false
+	for _, entry := range analysisEntries {
+		if entry.AnalyzedInstanceKey.Equals(&analysisEntry.AnalyzedInstanceKey) {
+			if entry.Analysis == inst.DeadPrimary || entry.Analysis == inst.DeadPrimaryAndSomeReplicas {
+				recoveryRequired = true
+			}
+		}
+	}
+
+	// No recovery is required. Some other agent already fixed the issue.
+	if !recoveryRequired {
+		log.Infof("Analysis: %v - No longer valid, some other agent must have fixed the problem.", analysisEntry.Analysis)
+		return false, nil, nil
+	}
+
+	// Read the tablet information from the database to find the shard and keyspace of the tablet
+	tablet, err := inst.ReadTablet(analysisEntry.AnalyzedInstanceKey)
 
 	var candidateTabletAlias *topodatapb.TabletAlias
 	if candidateInstanceKey != nil {
@@ -639,32 +672,6 @@ func checkAndRecoverDeadPrimary(analysisEntry inst.ReplicationAnalysis, candidat
 	}
 	log.Infof("Analysis: %v, deadprimary %+v", analysisEntry.Analysis, analysisEntry.AnalyzedInstanceKey)
 
-	// this check is needed because sometimes DeadPrimary code path is forcefully spawned off from other recoveries like PrimaryHasPrimary.
-	// So we need to check that we only run an ERS if the instance that we analyzed was actually a primary! Otherwise, we would end up running an ERS
-	// even when the cluster is fine or the problem can be fixed via some other recovery
-	if tablet.Type != topodatapb.TabletType_PRIMARY {
-		RefreshTablets(true /* forceRefresh */)
-		AuditTopologyRecovery(topologyRecovery, "another agent seems to have fixed the problem")
-		return false, topologyRecovery, nil
-	}
-
-	// check if we have received SIGTERM, if we have, we should not continue with the recovery
-	val := atomic.LoadInt32(&hasReceivedSIGTERM)
-	if val > 0 {
-		return false, topologyRecovery, errors.New("Can't lock shard: SIGTERM received")
-	}
-
-	// check if we have received an ERS in progress, if we do, we should not continue with the recovery
-	if checkAndSetIfERSInProgress() {
-		AuditTopologyRecovery(topologyRecovery, "an ERS is already in progress, not issuing another")
-		return false, topologyRecovery, nil
-	}
-	defer setERSCompleted()
-
-	// add to the shard lock counter since ERS will lock the shard
-	atomic.AddInt32(&shardsLockCounter, 1)
-	defer atomic.AddInt32(&shardsLockCounter, -1)
-
 	ev, err := reparentutil.NewEmergencyReparenter(ts, tmclient.NewTabletManagerClient(), logutil.NewCallbackLogger(func(event *logutilpb.Event) {
 		level := event.GetLevel()
 		value := event.GetValue()
@@ -676,7 +683,7 @@ func checkAndRecoverDeadPrimary(analysisEntry inst.ReplicationAnalysis, candidat
 			log.Errorf("ERS - %s", value)
 		}
 		AuditTopologyRecovery(topologyRecovery, value)
-	})).ReparentShard(context.Background(),
+	})).ReparentShard(ctx,
 		tablet.Keyspace,
 		tablet.Shard,
 		reparentutil.EmergencyReparentOptions{
@@ -687,9 +694,7 @@ func checkAndRecoverDeadPrimary(analysisEntry inst.ReplicationAnalysis, candidat
 		},
 	)
 
-	// here we need to forcefully refresh all the tablets otherwise old information is used and failover scenarios are spawned off which are not required
-	// For example, if we do not refresh the tablets forcefully and the new primary is found in the cache then its source key is not updated and this spawns off
-	// PrimaryHasPrimary analysis which runs another ERS
+	// We should refresh the tablet information again to update our information.
 	RefreshTablets(true /* forceRefresh */)
 	var promotedReplica *inst.Instance
 	if ev.NewPrimary != nil {
@@ -702,24 +707,6 @@ func checkAndRecoverDeadPrimary(analysisEntry inst.ReplicationAnalysis, candidat
 	return true, topologyRecovery, err
 }
 
-// checkAndSetIfERSInProgress checks if an ERS is already in progress. If it is not in progress, then we set it to be in progress.
-func checkAndSetIfERSInProgress() bool {
-	ersInProgressMutex.Lock()
-	defer ersInProgressMutex.Unlock()
-	if ersInProgress {
-		return true
-	}
-	ersInProgress = true
-	return false
-}
-
-// setERSCompleted sets the variable tracking if an ers is in progress to false.
-func setERSCompleted() {
-	ersInProgressMutex.Lock()
-	defer ersInProgressMutex.Unlock()
-	ersInProgress = false
-}
-
 func postErsCompletion(topologyRecovery *TopologyRecovery, analysisEntry inst.ReplicationAnalysis, skipProcesses bool, promotedReplica *inst.Instance) {
 	if promotedReplica != nil {
 		message := fmt.Sprintf("promoted replica: %+v", promotedReplica.Key)
@@ -1770,6 +1757,41 @@ func postPrsCompletion(topologyRecovery *TopologyRecovery, analysisEntry inst.Re
 
 // electNewPrimary elects a new primary while none were present before.
 func electNewPrimary(analysisEntry inst.ReplicationAnalysis, candidateInstanceKey *inst.InstanceKey, forceInstanceRecovery bool, skipProcesses bool) (recoveryAttempted bool, topologyRecovery *TopologyRecovery, err error) {
+	// We lock the shard here and then refresh the tablets information
+	ctx, unlock, err := LockShard(context.Background(), analysisEntry.AnalyzedInstanceKey)
+	if err != nil {
+		return false, nil, err
+	}
+	defer unlock(&err)
+
+	// TODO (@GuptaManan100): Refresh only the shard tablet information instead of all the tablets
+	RefreshTablets(true /* forceRefresh */)
+
+	// Run a replication analysis again. We need this because vtorc works on ephemeral data to find the failure scenarios.
+	// That data might be old, because of a cluster operation that was run through vtctld or some other vtorc. So before we do any
+	// changes, we should be checking that this failure is indeed needed to be fixed. We do this after locking the shard to be sure
+	// that the data that we use now is up-to-date.
+	analysisEntries, err := inst.GetReplicationAnalysis(analysisEntry.ClusterDetails.ClusterName, &inst.ReplicationAnalysisHints{})
+	if err != nil {
+		return false, nil, err
+	}
+
+	// The recovery is only required if the same instance key requires a ClusterHasNoPrimary recovery.
+	recoveryRequired := false
+	for _, entry := range analysisEntries {
+		if entry.AnalyzedInstanceKey.Equals(&analysisEntry.AnalyzedInstanceKey) {
+			if entry.Analysis == inst.ClusterHasNoPrimary {
+				recoveryRequired = true
+			}
+		}
+	}
+
+	// No recovery is required. Some other agent already fixed the issue.
+	if !recoveryRequired {
+		log.Infof("Analysis: %v - No longer valid, some other agent must have fixed the problem.", analysisEntry.Analysis)
+		return false, nil, nil
+	}
+
 	topologyRecovery, err = AttemptRecoveryRegistration(&analysisEntry, false /*failIfFailedInstanceInActiveRecovery*/, true /*failIfClusterInActiveRecovery*/)
 	if topologyRecovery == nil || err != nil {
 		AuditTopologyRecovery(topologyRecovery, fmt.Sprintf("found an active or recent recovery on %+v. Will not issue another electNewPrimary.", analysisEntry.AnalyzedInstanceKey))
@@ -1794,7 +1816,7 @@ func electNewPrimary(analysisEntry inst.ReplicationAnalysis, candidateInstanceKe
 			log.Errorf("PRS - %s", value)
 		}
 		AuditTopologyRecovery(topologyRecovery, value)
-	})).ReparentShard(context.Background(),
+	})).ReparentShard(ctx,
 		analyzedTablet.Keyspace,
 		analyzedTablet.Shard,
 		reparentutil.PlannedReparentOptions{
@@ -1819,6 +1841,46 @@ func electNewPrimary(analysisEntry inst.ReplicationAnalysis, candidateInstanceKe
 
 // fixClusterAndPrimary performs a traditional vitess PlannedReparentShard.
 func fixClusterAndPrimary(analysisEntry inst.ReplicationAnalysis, candidateInstanceKey *inst.InstanceKey, forceInstanceRecovery bool, skipProcesses bool) (recoveryAttempted bool, topologyRecovery *TopologyRecovery, err error) {
+	// We lock the shard here and then refresh the tablets information
+	_, unlock, err := LockShard(context.Background(), analysisEntry.AnalyzedInstanceKey)
+	if err != nil {
+		return false, nil, err
+	}
+	unlockFunctionCalled := false
+	defer func() {
+		if !unlockFunctionCalled {
+			unlock(&err)
+		}
+	}()
+
+	// TODO (@GuptaManan100): Refresh only the shard tablet information instead of all the tablets
+	RefreshTablets(true /* forceRefresh */)
+
+	// Run a replication analysis again. We need this because vtorc works on ephemeral data to find the failure scenarios.
+	// That data might be old, because of a cluster operation that was run through vtctld or some other vtorc. So before we do any
+	// changes, we should be checking that this failure is indeed needed to be fixed. We do this after locking the shard to be sure
+	// that the data that we use now is up-to-date.
+	analysisEntries, err := inst.GetReplicationAnalysis(analysisEntry.ClusterDetails.ClusterName, &inst.ReplicationAnalysisHints{})
+	if err != nil {
+		return false, nil, err
+	}
+
+	// The recovery is only required if the same instance key requires a DeadPrimary or DeadPrimaryAndSomeReplicas recovery.
+	recoveryRequired := false
+	for _, entry := range analysisEntries {
+		if entry.AnalyzedInstanceKey.Equals(&analysisEntry.AnalyzedInstanceKey) {
+			if entry.Analysis == inst.PrimaryHasPrimary {
+				recoveryRequired = true
+			}
+		}
+	}
+
+	// No recovery is required. Some other agent already fixed the issue.
+	if !recoveryRequired {
+		log.Infof("Analysis: %v - No longer valid, some other agent must have fixed the problem.", analysisEntry.Analysis)
+		return false, nil, nil
+	}
+
 	topologyRecovery, err = AttemptRecoveryRegistration(&analysisEntry, false, true)
 	if topologyRecovery == nil {
 		AuditTopologyRecovery(topologyRecovery, fmt.Sprintf("found an active or recent recovery on %+v. Will not issue another fixClusterAndPrimary.", analysisEntry.AnalyzedInstanceKey))
@@ -1833,6 +1895,8 @@ func fixClusterAndPrimary(analysisEntry inst.ReplicationAnalysis, candidateInsta
 		return false, topologyRecovery, err
 	}
 
+	unlockFunctionCalled = true
+	unlock(&err)
 	altAnalysis, err := forceAnalysisEntry(analysisEntry.ClusterDetails.ClusterName, inst.DeadPrimary, "", &analysisEntry.AnalyzedInstancePrimaryKey)
 	if err != nil {
 		return false, topologyRecovery, err
diff --git a/go/vt/vtctl/reparentutil/emergency_reparenter.go b/go/vt/vtctl/reparentutil/emergency_reparenter.go
index 46bea34edb..1f0746cabf 100644
--- a/go/vt/vtctl/reparentutil/emergency_reparenter.go
+++ b/go/vt/vtctl/reparentutil/emergency_reparenter.go
@@ -96,13 +96,17 @@ func NewEmergencyReparenter(ts *topo.Server, tmc tmclient.TabletManagerClient, l
 // ReparentShard performs the EmergencyReparentShard operation on the given
 // keyspace and shard.
 func (erp *EmergencyReparenter) ReparentShard(ctx context.Context, keyspace string, shard string, opts EmergencyReparentOptions) (*events.Reparent, error) {
-	// First step is to lock the shard for the given operation
-	opts.lockAction = erp.getLockAction(opts.NewPrimaryAlias)
-	ctx, unlock, err := erp.ts.LockShard(ctx, keyspace, shard, opts.lockAction)
-	if err != nil {
-		return nil, err
+	var err error
+	// First step is to lock the shard for the given operation, if not already locked
+	if err = topo.CheckShardLocked(ctx, keyspace, shard); err != nil {
+		var unlock func(*error)
+		opts.lockAction = erp.getLockAction(opts.NewPrimaryAlias)
+		ctx, unlock, err = erp.ts.LockShard(ctx, keyspace, shard, opts.lockAction)
+		if err != nil {
+			return nil, err
+		}
+		defer unlock(&err)
 	}
-	defer unlock(&err)
 
 	// dispatch success or failure of ERS
 	ev := &events.Reparent{}
diff --git a/go/vt/vtctl/reparentutil/planned_reparenter.go b/go/vt/vtctl/reparentutil/planned_reparenter.go
index 3b1dabeb05..46e351a8db 100644
--- a/go/vt/vtctl/reparentutil/planned_reparenter.go
+++ b/go/vt/vtctl/reparentutil/planned_reparenter.go
@@ -86,15 +86,17 @@ func NewPlannedReparenter(ts *topo.Server, tmc tmclient.TabletManagerClient, log
 // and shard. It will make the provided tablet the primary for the shard, when
 // both the current and desired primary are reachable and in a good state.
 func (pr *PlannedReparenter) ReparentShard(ctx context.Context, keyspace string, shard string, opts PlannedReparentOptions) (*events.Reparent, error) {
-	opts.lockAction = pr.getLockAction(opts)
-
-	ctx, unlock, err := pr.ts.LockShard(ctx, keyspace, shard, opts.lockAction)
-	if err != nil {
-		return nil, err
+	var err error
+	if err = topo.CheckShardLocked(ctx, keyspace, shard); err != nil {
+		var unlock func(*error)
+		opts.lockAction = pr.getLockAction(opts)
+		ctx, unlock, err = pr.ts.LockShard(ctx, keyspace, shard, opts.lockAction)
+		if err != nil {
+			return nil, err
+		}
+		defer unlock(&err)
 	}
 
-	defer unlock(&err)
-
 	if opts.NewPrimaryAlias == nil && opts.AvoidPrimaryAlias == nil {
 		shardInfo, err := pr.ts.GetShard(ctx, keyspace, shard)
 		if err != nil {
diff --git a/go/vt/vtgate/engine/gen4_compare_v3.go b/go/vt/vtgate/engine/gen4_compare_v3.go
index 1792e9de05..824128897c 100644
--- a/go/vt/vtgate/engine/gen4_compare_v3.go
+++ b/go/vt/vtgate/engine/gen4_compare_v3.go
@@ -18,6 +18,7 @@ package engine
 
 import (
 	"encoding/json"
+	"sync"
 
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/log"
@@ -97,17 +98,22 @@ func (gc *Gen4CompareV3) TryStreamExecute(
 	wantfields bool,
 	callback func(*sqltypes.Result) error,
 ) error {
+	var mu sync.Mutex
 	var v3Err, gen4Err error
 	v3Result, gen4Result := &sqltypes.Result{}, &sqltypes.Result{}
 
 	if gc.Gen4 != nil {
 		gen4Err = gc.Gen4.TryStreamExecute(vcursor, bindVars, wantfields, func(result *sqltypes.Result) error {
+			mu.Lock()
+			defer mu.Unlock()
 			gen4Result.AppendResult(result)
 			return nil
 		})
 	}
 	if gc.V3 != nil {
 		v3Err = gc.V3.TryStreamExecute(vcursor, bindVars, wantfields, func(result *sqltypes.Result) error {
+			mu.Lock()
+			defer mu.Unlock()
 			v3Result.AppendResult(result)
 			return nil
 		})
diff --git a/go/vt/vtgate/planbuilder/aggregation_pushing.go b/go/vt/vtgate/planbuilder/aggregation_pushing.go
index 75997117f2..4707c667b6 100644
--- a/go/vt/vtgate/planbuilder/aggregation_pushing.go
+++ b/go/vt/vtgate/planbuilder/aggregation_pushing.go
@@ -34,23 +34,30 @@ func (hp *horizonPlanning) pushAggregation(
 	grouping []abstract.GroupBy,
 	aggregations []abstract.Aggr,
 	ignoreOutputOrder bool,
-) (groupingOffsets []offsets, outputAggrsOffset [][]offsets, err error) {
+) (output logicalPlan, groupingOffsets []offsets, outputAggrsOffset [][]offsets, err error) {
 	switch plan := plan.(type) {
 	case *routeGen4:
+		output = plan
 		groupingOffsets, outputAggrsOffset, err = pushAggrOnRoute(ctx, plan, aggregations, grouping, ignoreOutputOrder)
-		if err != nil {
-			return nil, nil, err
-		}
 		return
 
 	case *joinGen4:
-		return hp.pushAggrOnJoin(ctx, grouping, aggregations, plan)
+		output = plan
+		groupingOffsets, outputAggrsOffset, err = hp.pushAggrOnJoin(ctx, plan, grouping, aggregations)
+		return
 
 	case *semiJoin:
-		return hp.pushAggrOnSemiJoin(ctx, grouping, aggregations, plan, ignoreOutputOrder)
+		output = plan
+		groupingOffsets, outputAggrsOffset, err = hp.pushAggrOnSemiJoin(ctx, plan, grouping, aggregations, ignoreOutputOrder)
+		return
+
+	case *simpleProjection:
+		// we just remove the simpleProjection. We are doing an OA on top anyway, so no need to clean up the output columns
+		return hp.pushAggregation(ctx, plan.input, grouping, aggregations, ignoreOutputOrder)
 
 	default:
-		return nil, nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "using aggregation on top of a %T plan is not yet supported", plan)
+		err = vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "using aggregation on top of a %T plan is not yet supported", plan)
+		return
 	}
 }
 
@@ -195,9 +202,9 @@ vtgate level, we can offload most of the work to MySQL, and at the vtgate just s
 */
 func (hp *horizonPlanning) pushAggrOnJoin(
 	ctx *plancontext.PlanningContext,
+	join *joinGen4,
 	grouping []abstract.GroupBy,
 	aggregations []abstract.Aggr,
-	join *joinGen4,
 ) ([]offsets, [][]offsets, error) {
 	// First we separate aggregations according to which side the dependencies are coming from
 	lhsAggrs, rhsAggrs, err := splitAggregationsToLeftAndRight(ctx, aggregations, join)
@@ -220,15 +227,16 @@ func (hp *horizonPlanning) pushAggrOnJoin(
 	}
 
 	// Next we push the aggregations to both sides
-	lhsOffsets, lhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Left, lhsGrouping, lhsAggrs, true)
+	newLHS, lhsOffsets, lhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Left, lhsGrouping, lhsAggrs, true)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	rhsOffsets, rhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Right, rhsGrouping, rhsAggrs, true)
+	newRHS, rhsOffsets, rhsAggrOffsets, err := hp.filteredPushAggregation(ctx, join.Right, rhsGrouping, rhsAggrs, true)
 	if err != nil {
 		return nil, nil, err
 	}
+	join.Left, join.Right = newLHS, newRHS
 
 	// Next, we have to pass through the grouping values through the join and the projection we add on top
 	// We added new groupings to the LHS because of the join condition, so we don't want to pass through everything,
@@ -283,9 +291,9 @@ That way we get the aggregation grouped by the column we need to use to decide i
 */
 func (hp *horizonPlanning) pushAggrOnSemiJoin(
 	ctx *plancontext.PlanningContext,
+	join *semiJoin,
 	grouping []abstract.GroupBy,
 	aggregations []abstract.Aggr,
-	join *semiJoin,
 	ignoreOutputOrder bool,
 ) ([]offsets, [][]offsets, error) {
 	// We need to group by the columns used in the join condition.
@@ -296,10 +304,11 @@ func (hp *horizonPlanning) pushAggrOnSemiJoin(
 	}
 
 	totalGrouping := append(grouping, lhsCols...)
-	groupingOffsets, aggrParams, err := hp.pushAggregation(ctx, join.lhs, totalGrouping, aggregations, ignoreOutputOrder)
+	newLeft, groupingOffsets, aggrParams, err := hp.pushAggregation(ctx, join.lhs, totalGrouping, aggregations, ignoreOutputOrder)
 	if err != nil {
 		return nil, nil, err
 	}
+	join.lhs = newLeft
 
 	outputGroupings := make([]offsets, 0, len(grouping))
 	for idx := range grouping {
@@ -321,7 +330,7 @@ func (hp *horizonPlanning) filteredPushAggregation(
 	grouping []abstract.GroupBy,
 	aggregations []*abstract.Aggr,
 	ignoreOutputOrder bool,
-) (groupingOffsets []offsets, outputAggrs [][]offsets, err error) {
+) (out logicalPlan, groupingOffsets []offsets, outputAggrs [][]offsets, err error) {
 	used := make([]bool, len(aggregations))
 	var aggrs []abstract.Aggr
 
@@ -331,9 +340,9 @@ func (hp *horizonPlanning) filteredPushAggregation(
 			aggrs = append(aggrs, *aggr)
 		}
 	}
-	groupingOffsets, pushedAggrs, err := hp.pushAggregation(ctx, plan, grouping, aggrs, ignoreOutputOrder)
+	newplan, groupingOffsets, pushedAggrs, err := hp.pushAggregation(ctx, plan, grouping, aggrs, ignoreOutputOrder)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 	idx := 0
 	for _, b := range used {
@@ -344,7 +353,7 @@ func (hp *horizonPlanning) filteredPushAggregation(
 		outputAggrs = append(outputAggrs, pushedAggrs[idx])
 		idx++
 	}
-	return groupingOffsets, outputAggrs, nil
+	return newplan, groupingOffsets, outputAggrs, nil
 }
 
 func isMinOrMax(in engine.AggregateOpcode) bool {
diff --git a/go/vt/vtgate/planbuilder/horizon_planning.go b/go/vt/vtgate/planbuilder/horizon_planning.go
index f31ed95048..7831c13994 100644
--- a/go/vt/vtgate/planbuilder/horizon_planning.go
+++ b/go/vt/vtgate/planbuilder/horizon_planning.go
@@ -551,11 +551,13 @@ func (hp *horizonPlanning) planAggrUsingOA(
 		oa.preProcess = true
 	}
 
-	groupingOffsets, aggrParamOffsets, err := hp.pushAggregation(ctx, plan, grouping, aggrs, false)
+	newPlan, groupingOffsets, aggrParamOffsets, err := hp.pushAggregation(ctx, plan, grouping, aggrs, false)
 	if err != nil {
 		return nil, err
 	}
 
+	plan = newPlan
+
 	_, isRoute := plan.(*routeGen4)
 	needsProj := !isRoute
 	var aggPlan = plan
diff --git a/go/vt/zkctl/zksrv.sh b/go/vt/zkctl/zksrv.sh
index 26479b373c..9c7c6bd8c9 100755
--- a/go/vt/zkctl/zksrv.sh
+++ b/go/vt/zkctl/zksrv.sh
@@ -21,7 +21,7 @@ logdir="$1"
 config="$2"
 pidfile="$3"
 zk_java_opts=${ZK_JAVA_OPTS:-}
-zk_ver=${ZK_VERSION:-3.4.14}
+zk_ver=${ZK_VERSION:-3.8.0}
 classpath="$VTROOT/dist/vt-zookeeper-$zk_ver/lib/zookeeper-$zk_ver-fatjar.jar:/usr/local/lib/zookeeper-$zk_ver-fatjar.jar:/usr/share/java/zookeeper-$zk_ver.jar"
 
 mkdir -p "$logdir"
diff --git a/tools/shell_functions.inc b/tools/shell_functions.inc
index 67b299e314..a00ed7df41 100644
--- a/tools/shell_functions.inc
+++ b/tools/shell_functions.inc
@@ -26,11 +26,11 @@ function goversion_min() {
   wantmajor=${BASH_REMATCH[1]}
   wantminor=${BASH_REMATCH[2]}
   wantpatch=${BASH_REMATCH[3]}
-  [ "$gotmajor" -lt "$wantmajor" ] && return 1
-  [ "$gotmajor" -gt "$wantmajor" ] && return 0
-  [ "$gotminor" -lt "$wantminor" ] && return 1
-  [ "$gotminor" -gt "$wantminor" ] && return 0
-  [ "$gotpatch" -lt "$wantpatch" ] && return 1
+  [[ $gotmajor -lt $wantmajor ]] && return 1
+  [[ $gotmajor -gt $wantmajor ]] && return 0
+  [[ $gotminor -lt $wantminor ]] && return 1
+  [[ $gotminor -gt $wantminor ]] && return 0
+  [[ $gotpatch -lt $wantpatch ]] && return 1
   return 0
 }
 


Test patch:
diff --git a/.github/docker/cluster_test_12/Dockerfile b/.github/docker/cluster_test_12/Dockerfile
index 604b2b5165..482602e479 100644
--- a/.github/docker/cluster_test_12/Dockerfile
+++ b/.github/docker/cluster_test_12/Dockerfile
@@ -1,6 +1,6 @@
 # DO NOT MODIFY: THIS FILE IS GENERATED USING "make generate_ci_workflows"
 
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql57"
 
 FROM "${image}"
diff --git a/.github/docker/cluster_test_18/Dockerfile b/.github/docker/cluster_test_18/Dockerfile
index d19aeb56b7..803ad3dfe0 100644
--- a/.github/docker/cluster_test_18/Dockerfile
+++ b/.github/docker/cluster_test_18/Dockerfile
@@ -1,6 +1,6 @@
 # DO NOT MODIFY: THIS FILE IS GENERATED USING "make generate_ci_workflows"
 
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql57"
 
 FROM "${image}"
diff --git a/docker/lite/Dockerfile.testing b/docker/lite/Dockerfile.testing
index 6a92e59ef6..3db75247d2 100644
--- a/docker/lite/Dockerfile.testing
+++ b/docker/lite/Dockerfile.testing
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql57"
 
 FROM "${image}" AS builder
diff --git a/docker/vttestserver/Dockerfile.mysql57 b/docker/vttestserver/Dockerfile.mysql57
index 0276ba497f..9099267895 100644
--- a/docker/vttestserver/Dockerfile.mysql57
+++ b/docker/vttestserver/Dockerfile.mysql57
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql57"
 
 FROM "${image}" AS builder
diff --git a/docker/vttestserver/Dockerfile.mysql80 b/docker/vttestserver/Dockerfile.mysql80
index 286e0ebb8a..f9d497f129 100644
--- a/docker/vttestserver/Dockerfile.mysql80
+++ b/docker/vttestserver/Dockerfile.mysql80
@@ -17,7 +17,7 @@
 #       ensure images contain the right binaries.
 
 # Use a temporary layer for the build stage.
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-mysql80"
 
 FROM "${image}" AS builder
diff --git a/go/test/endtoend/cluster/cluster_process.go b/go/test/endtoend/cluster/cluster_process.go
index 391e98f271..f6cec182f7 100644
--- a/go/test/endtoend/cluster/cluster_process.go
+++ b/go/test/endtoend/cluster/cluster_process.go
@@ -89,7 +89,7 @@ type LocalProcessCluster struct {
 	VtgateProcess   VtgateProcess
 	VtworkerProcess VtworkerProcess
 	VtbackupProcess VtbackupProcess
-	VtorcProcess    *VtorcProcess
+	VtorcProcesses  []*VtorcProcess
 
 	nextPortForProcess int
 
@@ -726,8 +726,8 @@ func (cluster *LocalProcessCluster) Teardown() {
 		log.Errorf("Error in vtgate teardown: %v", err)
 	}
 
-	if cluster.VtorcProcess != nil {
-		if err := cluster.VtorcProcess.TearDown(); err != nil {
+	for _, vtorcProcess := range cluster.VtorcProcesses {
+		if err := vtorcProcess.TearDown(); err != nil {
 			log.Errorf("Error in vtorc teardown: %v", err)
 		}
 	}
@@ -952,13 +952,14 @@ func (cluster *LocalProcessCluster) NewVttabletInstance(tabletType string, UID i
 }
 
 // NewOrcProcess creates a new VtorcProcess object
-func (cluster *LocalProcessCluster) NewOrcProcess(configFile string) *VtorcProcess {
+func (cluster *LocalProcessCluster) NewOrcProcess(config VtorcConfiguration) *VtorcProcess {
 	base := VtctlProcessInstance(cluster.TopoProcess.Port, cluster.Hostname)
 	base.Binary = "vtorc"
 	return &VtorcProcess{
 		VtctlProcess: *base,
 		LogDir:       cluster.TmpDirectory,
-		Config:       configFile,
+		Config:       config,
+		WebPort:      cluster.GetAndReservePort(),
 	}
 }
 
diff --git a/go/test/endtoend/cluster/vtorc_process.go b/go/test/endtoend/cluster/vtorc_process.go
index fc5e6264f8..21f743f975 100644
--- a/go/test/endtoend/cluster/vtorc_process.go
+++ b/go/test/endtoend/cluster/vtorc_process.go
@@ -18,6 +18,7 @@ limitations under the License.
 package cluster
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
 	"os/exec"
@@ -33,16 +34,68 @@ import (
 // vtorc as a separate process for testing
 type VtorcProcess struct {
 	VtctlProcess
-	LogDir    string
-	ExtraArgs []string
-	Config    string
-	proc      *exec.Cmd
-	exit      chan error
+	LogDir     string
+	ExtraArgs  []string
+	ConfigPath string
+	Config     VtorcConfiguration
+	WebPort    int
+	proc       *exec.Cmd
+	exit       chan error
+}
+
+type VtorcConfiguration struct {
+	Debug                                 bool
+	ListenAddress                         string
+	MySQLTopologyUser                     string
+	MySQLTopologyPassword                 string
+	MySQLReplicaUser                      string
+	MySQLReplicaPassword                  string
+	RecoveryPeriodBlockSeconds            int
+	InstancePollSeconds                   int
+	PreventCrossDataCenterPrimaryFailover bool   `json:",omitempty"`
+	LockShardTimeoutSeconds               int    `json:",omitempty"`
+	Durability                            string `json:",omitempty"`
+	ReplicationLagQuery                   string `json:",omitempty"`
+	FailPrimaryPromotionOnLagMinutes      int    `json:",omitempty"`
+}
+
+// ToJSONString will marshal this configuration as JSON
+func (config *VtorcConfiguration) ToJSONString() string {
+	b, _ := json.MarshalIndent(config, "", "\t")
+	return string(b)
+}
+
+func (config *VtorcConfiguration) AddDefaults(webPort int) {
+	config.Debug = true
+	config.MySQLTopologyUser = "orc_client_user"
+	config.MySQLTopologyPassword = "orc_client_user_password"
+	config.MySQLReplicaUser = "vt_repl"
+	config.MySQLReplicaPassword = ""
+	config.RecoveryPeriodBlockSeconds = 1
+	config.InstancePollSeconds = 1
+	config.ListenAddress = fmt.Sprintf(":%d", webPort)
 }
 
 // Setup starts orc process with required arguements
 func (orc *VtorcProcess) Setup() (err error) {
 
+	// create the configuration file
+	timeNow := time.Now().UnixNano()
+	configFile, _ := os.Create(path.Join(orc.LogDir, fmt.Sprintf("orc-config-%d.json", timeNow)))
+	orc.ConfigPath = configFile.Name()
+
+	// Add the default configurations and print them out
+	orc.Config.AddDefaults(orc.WebPort)
+	log.Errorf("configuration - %v", orc.Config.ToJSONString())
+	_, err = configFile.WriteString(orc.Config.ToJSONString())
+	if err != nil {
+		return err
+	}
+	err = configFile.Close()
+	if err != nil {
+		return err
+	}
+
 	/* minimal command line arguments:
 	$ vtorc -topo_implementation etcd2 -topo_global_server_address localhost:2379 -topo_global_root /vitess/global
 	-config config/orchestrator/default.json -alsologtostderr http
@@ -52,7 +105,7 @@ func (orc *VtorcProcess) Setup() (err error) {
 		"--topo_implementation", orc.TopoImplementation,
 		"--topo_global_server_address", orc.TopoGlobalAddress,
 		"--topo_global_root", orc.TopoGlobalRoot,
-		"--config", orc.Config,
+		"--config", orc.ConfigPath,
 		"--orc_web_dir", path.Join(os.Getenv("VTROOT"), "web", "orchestrator"),
 	)
 	if *isCoverage {
@@ -62,7 +115,7 @@ func (orc *VtorcProcess) Setup() (err error) {
 	orc.proc.Args = append(orc.proc.Args, orc.ExtraArgs...)
 	orc.proc.Args = append(orc.proc.Args, "--alsologtostderr", "http")
 
-	errFile, _ := os.Create(path.Join(orc.LogDir, fmt.Sprintf("orc-stderr-%d.txt", time.Now().UnixNano())))
+	errFile, _ := os.Create(path.Join(orc.LogDir, fmt.Sprintf("orc-stderr-%d.txt", timeNow)))
 	orc.proc.Stderr = errFile
 
 	orc.proc.Env = append(orc.proc.Env, os.Environ()...)
diff --git a/go/test/endtoend/reparent/emergencyreparent/ers_test.go b/go/test/endtoend/reparent/emergencyreparent/ers_test.go
index 34e76e8987..82d51b493a 100644
--- a/go/test/endtoend/reparent/emergencyreparent/ers_test.go
+++ b/go/test/endtoend/reparent/emergencyreparent/ers_test.go
@@ -18,9 +18,11 @@ package emergencyreparent
 
 import (
 	"context"
+	"os/exec"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/test/endtoend/cluster"
@@ -223,3 +225,220 @@ func TestSemiSyncSetupCorrectly(t *testing.T) {
 		}
 	})
 }
+
+// TestERSPromoteRdonly tests that we never end up promoting a rdonly instance as the primary
+func TestERSPromoteRdonly(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
+	defer utils.TeardownCluster(clusterInstance)
+	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+	var err error
+
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", tablets[1].Alias, "rdonly")
+	require.NoError(t, err)
+
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", tablets[2].Alias, "rdonly")
+	require.NoError(t, err)
+
+	utils.ConfirmReplication(t, tablets[0], tablets[1:])
+
+	// Make the current primary agent and database unavailable.
+	utils.StopTablet(t, tablets[0], true)
+
+	// We expect this one to fail because we have ignored all the replicas and have only the rdonly's which should not be promoted
+	out, err := utils.ErsIgnoreTablet(clusterInstance, nil, "30s", "30s", []*cluster.Vttablet{tablets[3]}, false)
+	require.NotNil(t, err, out)
+
+	out, err = clusterInstance.VtctlclientProcess.ExecuteCommandWithOutput("GetShard", utils.KeyspaceShard)
+	require.NoError(t, err)
+	require.Contains(t, out, `"uid": 101`, "the primary should still be 101 in the shard info")
+}
+
+// TestERSPreventCrossCellPromotion tests that we promote a replica in the same cell as the previous primary if prevent cross cell promotion flag is set
+func TestERSPreventCrossCellPromotion(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
+	defer utils.TeardownCluster(clusterInstance)
+	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+	var err error
+
+	// confirm that replication is going smoothly
+	utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
+
+	// Make the current primary agent and database unavailable.
+	utils.StopTablet(t, tablets[0], true)
+
+	// We expect that tablets[2] will be promoted since it is in the same cell as the previous primary
+	out, err := utils.ErsIgnoreTablet(clusterInstance, nil, "60s", "30s", []*cluster.Vttablet{tablets[1]}, true)
+	require.NoError(t, err, out)
+
+	newPrimary := utils.GetNewPrimary(t, clusterInstance)
+	require.Equal(t, newPrimary.Alias, tablets[2].Alias, "tablets[2] should be the promoted primary")
+}
+
+// TestPullFromRdonly tests that if a rdonly tablet is the most advanced, then our promoted primary should have
+// caught up to it by pulling transactions from it
+func TestPullFromRdonly(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
+	defer utils.TeardownCluster(clusterInstance)
+	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+	var err error
+
+	ctx := context.Background()
+	// make tablets[1] a rdonly tablet.
+	// rename tablet so that the test is not confusing
+	rdonly := tablets[1]
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", rdonly.Alias, "rdonly")
+	require.NoError(t, err)
+
+	// confirm that all the tablets can replicate successfully right now
+	utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{rdonly, tablets[2], tablets[3]})
+
+	// stop replication on the other two tablets
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StopReplication", tablets[2].Alias)
+	require.NoError(t, err)
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StopReplication", tablets[3].Alias)
+	require.NoError(t, err)
+
+	// stop semi-sync on the primary so that any transaction now added does not require an ack
+	utils.RunSQL(ctx, t, "SET GLOBAL rpl_semi_sync_master_enabled = false", tablets[0])
+
+	// confirm that rdonly is able to replicate from our primary
+	// This will also introduce a new transaction into the rdonly tablet which the other 2 replicas don't have
+	insertVal := utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{rdonly})
+
+	// Make the current primary agent and database unavailable.
+	utils.StopTablet(t, tablets[0], true)
+
+	// start the replication back on the two tablets
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StartReplication", tablets[2].Alias)
+	require.NoError(t, err)
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StartReplication", tablets[3].Alias)
+	require.NoError(t, err)
+
+	// check that tablets[2] and tablets[3] still only has 1 value
+	err = utils.CheckCountOfInsertedValues(ctx, t, tablets[2], 1)
+	require.NoError(t, err)
+	err = utils.CheckCountOfInsertedValues(ctx, t, tablets[3], 1)
+	require.NoError(t, err)
+
+	// At this point we have successfully made our rdonly tablet more advanced than tablets[2] and tablets[3] without introducing errant GTIDs
+	// We have simulated a network partition in which the primary and rdonly got isolated and then the primary went down leaving the rdonly most advanced
+
+	// We expect that tablets[2] will be promoted since it is in the same cell as the previous primary
+	// since we are preventing cross cell promotions
+	// Also it must be fully caught up
+	out, err := utils.ErsIgnoreTablet(clusterInstance, nil, "60s", "30s", nil, true)
+	require.NoError(t, err, out)
+
+	newPrimary := utils.GetNewPrimary(t, clusterInstance)
+	require.Equal(t, newPrimary.Alias, tablets[2].Alias, "tablets[2] should be the promoted primary")
+
+	// check that the new primary has the last transaction that only the rdonly had
+	err = utils.CheckInsertedValues(ctx, t, newPrimary, insertVal)
+	require.NoError(t, err)
+}
+
+// TestNoReplicationStatusAndReplicationStopped checks that ERS is able to fix
+// replicas which do not have any replication status and also succeeds if the replication
+// is stopped on the primary elect.
+func TestNoReplicationStatusAndReplicationStopped(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
+	defer utils.TeardownCluster(clusterInstance)
+	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+	utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
+
+	err := clusterInstance.VtctlclientProcess.ExecuteCommand("ExecuteFetchAsDba", tablets[1].Alias, `STOP SLAVE; RESET SLAVE ALL`)
+	require.NoError(t, err)
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ExecuteFetchAsDba", tablets[2].Alias, `STOP SLAVE;`)
+	require.NoError(t, err)
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ExecuteFetchAsDba", tablets[3].Alias, `STOP SLAVE SQL_THREAD;`)
+	require.NoError(t, err)
+	// Run an additional command in the current primary which will only be acked by tablets[3] and be in its relay log.
+	insertedVal := utils.ConfirmReplication(t, tablets[0], nil)
+	// Failover to tablets[3]
+	out, err := utils.Ers(clusterInstance, tablets[3], "60s", "30s")
+	require.NoError(t, err, out)
+	// Verify that the tablet has the inserted value
+	err = utils.CheckInsertedValues(context.Background(), t, tablets[3], insertedVal)
+	require.NoError(t, err)
+	// Confirm that replication is setup correctly from tablets[3] to tablets[0]
+	utils.ConfirmReplication(t, tablets[3], tablets[:1])
+	// Confirm that tablets[2] which had replication stopped initially still has its replication stopped
+	utils.CheckReplicationStatus(context.Background(), t, tablets[2], false, false)
+}
+
+// TestERSForInitialization tests whether calling ERS in the beginning sets up the cluster properly or not
+func TestERSForInitialization(t *testing.T) {
+	var tablets []*cluster.Vttablet
+	clusterInstance := cluster.NewCluster("zone1", "localhost")
+	defer clusterInstance.Teardown()
+	keyspace := &cluster.Keyspace{Name: utils.KeyspaceName}
+	clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "--durability_policy=semi_sync")
+	// Start topo server
+	err := clusterInstance.StartTopo()
+	require.NoError(t, err)
+	err = clusterInstance.TopoProcess.ManageTopoDir("mkdir", "/vitess/"+"zone1")
+	require.NoError(t, err)
+	for i := 0; i < 4; i++ {
+		tablet := clusterInstance.NewVttabletInstance("replica", 100+i, "zone1")
+		tablets = append(tablets, tablet)
+	}
+
+	shard := &cluster.Shard{Name: utils.ShardName}
+	shard.Vttablets = tablets
+	clusterInstance.VtTabletExtraArgs = []string{
+		"--lock_tables_timeout", "5s",
+		"--enable_semi_sync",
+		"--init_populate_metadata",
+		"--track_schema_versions=true",
+	}
+
+	// Initialize Cluster
+	err = clusterInstance.SetupCluster(keyspace, []cluster.Shard{*shard})
+	require.NoError(t, err)
+
+	//Start MySql
+	var mysqlCtlProcessList []*exec.Cmd
+	for _, shard := range clusterInstance.Keyspaces[0].Shards {
+		for _, tablet := range shard.Vttablets {
+			log.Infof("Starting MySql for tablet %v", tablet.Alias)
+			proc, err := tablet.MysqlctlProcess.StartProcess()
+			require.NoError(t, err)
+			mysqlCtlProcessList = append(mysqlCtlProcessList, proc)
+		}
+	}
+	// Wait for mysql processes to start
+	for _, proc := range mysqlCtlProcessList {
+		if err := proc.Wait(); err != nil {
+			t.Fatalf("Error starting mysql: %s", err.Error())
+		}
+	}
+
+	for _, tablet := range tablets {
+		// Start the tablet
+		err = tablet.VttabletProcess.Setup()
+		require.NoError(t, err)
+	}
+	for _, tablet := range tablets {
+		err := tablet.VttabletProcess.WaitForTabletStatuses([]string{"SERVING", "NOT_SERVING"})
+		require.NoError(t, err)
+	}
+
+	// Force the replica to reparent assuming that all the datasets are identical.
+	res, err := utils.Ers(clusterInstance, tablets[0], "60s", "30s")
+	require.NoError(t, err, res)
+
+	utils.ValidateTopology(t, clusterInstance, true)
+	// create Tables
+	utils.RunSQL(context.Background(), t, "create table vt_insert_test (id bigint, msg varchar(64), primary key (id)) Engine=InnoDB", tablets[0])
+	utils.CheckPrimaryTablet(t, clusterInstance, tablets[0])
+	utils.ValidateTopology(t, clusterInstance, false)
+	time.Sleep(100 * time.Millisecond) // wait for replication to catchup
+	strArray := utils.GetShardReplicationPositions(t, clusterInstance, utils.KeyspaceName, utils.ShardName, true)
+	assert.Equal(t, len(tablets), len(strArray))
+	assert.Contains(t, strArray[0], "primary") // primary first
+	utils.ConfirmReplication(t, tablets[0], tablets[1:])
+}
diff --git a/go/test/endtoend/reparent/newfeaturetest/reparent_test.go b/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
index 4d6286ce4f..025ac6319c 100644
--- a/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
+++ b/go/test/endtoend/reparent/newfeaturetest/reparent_test.go
@@ -17,238 +17,17 @@ limitations under the License.
 package newfeaturetest
 
 import (
-	"context"
-	"os/exec"
 	"testing"
 	"time"
 
 	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/test/endtoend/reparent/utils"
-	"vitess.io/vitess/go/vt/log"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 // ERS TESTS
 
-// TestERSPromoteRdonly tests that we never end up promoting a rdonly instance as the primary
-func TestERSPromoteRdonly(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t, true)
-	defer utils.TeardownCluster(clusterInstance)
-	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
-	var err error
-
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", tablets[1].Alias, "rdonly")
-	require.NoError(t, err)
-
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", tablets[2].Alias, "rdonly")
-	require.NoError(t, err)
-
-	utils.ConfirmReplication(t, tablets[0], tablets[1:])
-
-	// Make the current primary agent and database unavailable.
-	utils.StopTablet(t, tablets[0], true)
-
-	// We expect this one to fail because we have ignored all the replicas and have only the rdonly's which should not be promoted
-	out, err := utils.ErsIgnoreTablet(clusterInstance, nil, "30s", "30s", []*cluster.Vttablet{tablets[3]}, false)
-	require.NotNil(t, err, out)
-
-	out, err = clusterInstance.VtctlclientProcess.ExecuteCommandWithOutput("GetShard", utils.KeyspaceShard)
-	require.NoError(t, err)
-	require.Contains(t, out, `"uid": 101`, "the primary should still be 101 in the shard info")
-}
-
-// TestERSPreventCrossCellPromotion tests that we promote a replica in the same cell as the previous primary if prevent cross cell promotion flag is set
-func TestERSPreventCrossCellPromotion(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t, true)
-	defer utils.TeardownCluster(clusterInstance)
-	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
-	var err error
-
-	// confirm that replication is going smoothly
-	utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
-
-	// Make the current primary agent and database unavailable.
-	utils.StopTablet(t, tablets[0], true)
-
-	// We expect that tablets[2] will be promoted since it is in the same cell as the previous primary
-	out, err := utils.ErsIgnoreTablet(clusterInstance, nil, "60s", "30s", []*cluster.Vttablet{tablets[1]}, true)
-	require.NoError(t, err, out)
-
-	newPrimary := utils.GetNewPrimary(t, clusterInstance)
-	require.Equal(t, newPrimary.Alias, tablets[2].Alias, "tablets[2] should be the promoted primary")
-}
-
-// TestPullFromRdonly tests that if a rdonly tablet is the most advanced, then our promoted primary should have
-// caught up to it by pulling transactions from it
-func TestPullFromRdonly(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t, true)
-	defer utils.TeardownCluster(clusterInstance)
-	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
-	var err error
-
-	ctx := context.Background()
-	// make tablets[1] a rdonly tablet.
-	// rename tablet so that the test is not confusing
-	rdonly := tablets[1]
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", rdonly.Alias, "rdonly")
-	require.NoError(t, err)
-
-	// confirm that all the tablets can replicate successfully right now
-	utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{rdonly, tablets[2], tablets[3]})
-
-	// stop replication on the other two tablets
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StopReplication", tablets[2].Alias)
-	require.NoError(t, err)
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StopReplication", tablets[3].Alias)
-	require.NoError(t, err)
-
-	// stop semi-sync on the primary so that any transaction now added does not require an ack
-	utils.RunSQL(ctx, t, "SET GLOBAL rpl_semi_sync_master_enabled = false", tablets[0])
-
-	// confirm that rdonly is able to replicate from our primary
-	// This will also introduce a new transaction into the rdonly tablet which the other 2 replicas don't have
-	insertVal := utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{rdonly})
-
-	// Make the current primary agent and database unavailable.
-	utils.StopTablet(t, tablets[0], true)
-
-	// start the replication back on the two tablets
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StartReplication", tablets[2].Alias)
-	require.NoError(t, err)
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StartReplication", tablets[3].Alias)
-	require.NoError(t, err)
-
-	// check that tablets[2] and tablets[3] still only has 1 value
-	err = utils.CheckCountOfInsertedValues(ctx, t, tablets[2], 1)
-	require.NoError(t, err)
-	err = utils.CheckCountOfInsertedValues(ctx, t, tablets[3], 1)
-	require.NoError(t, err)
-
-	// At this point we have successfully made our rdonly tablet more advanced than tablets[2] and tablets[3] without introducing errant GTIDs
-	// We have simulated a network partition in which the primary and rdonly got isolated and then the primary went down leaving the rdonly most advanced
-
-	// We expect that tablets[2] will be promoted since it is in the same cell as the previous primary
-	// since we are preventing cross cell promotions
-	// Also it must be fully caught up
-	out, err := utils.ErsIgnoreTablet(clusterInstance, nil, "60s", "30s", nil, true)
-	require.NoError(t, err, out)
-
-	newPrimary := utils.GetNewPrimary(t, clusterInstance)
-	require.Equal(t, newPrimary.Alias, tablets[2].Alias, "tablets[2] should be the promoted primary")
-
-	// check that the new primary has the last transaction that only the rdonly had
-	err = utils.CheckInsertedValues(ctx, t, newPrimary, insertVal)
-	require.NoError(t, err)
-}
-
-// TestNoReplicationStatusAndReplicationStopped checks that ERS is able to fix
-// replicas which do not have any replication status and also succeeds if the replication
-// is stopped on the primary elect.
-func TestNoReplicationStatusAndReplicationStopped(t *testing.T) {
-	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t, true)
-	defer utils.TeardownCluster(clusterInstance)
-	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
-	utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
-
-	err := clusterInstance.VtctlclientProcess.ExecuteCommand("ExecuteFetchAsDba", tablets[1].Alias, `STOP SLAVE; RESET SLAVE ALL`)
-	require.NoError(t, err)
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ExecuteFetchAsDba", tablets[2].Alias, `STOP SLAVE;`)
-	require.NoError(t, err)
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ExecuteFetchAsDba", tablets[3].Alias, `STOP SLAVE SQL_THREAD;`)
-	require.NoError(t, err)
-	// Run an additional command in the current primary which will only be acked by tablets[3] and be in its relay log.
-	insertedVal := utils.ConfirmReplication(t, tablets[0], nil)
-	// Failover to tablets[3]
-	out, err := utils.Ers(clusterInstance, tablets[3], "60s", "30s")
-	require.NoError(t, err, out)
-	// Verify that the tablet has the inserted value
-	err = utils.CheckInsertedValues(context.Background(), t, tablets[3], insertedVal)
-	require.NoError(t, err)
-	// Confirm that replication is setup correctly from tablets[3] to tablets[0]
-	utils.ConfirmReplication(t, tablets[3], tablets[:1])
-	// Confirm that tablets[2] which had replication stopped initially still has its replication stopped
-	utils.CheckReplicationStatus(context.Background(), t, tablets[2], false, false)
-}
-
-// TestERSForInitialization tests whether calling ERS in the beginning sets up the cluster properly or not
-func TestERSForInitialization(t *testing.T) {
-	var tablets []*cluster.Vttablet
-	clusterInstance := cluster.NewCluster("zone1", "localhost")
-	defer clusterInstance.Teardown()
-	keyspace := &cluster.Keyspace{Name: utils.KeyspaceName}
-	clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "--durability_policy=semi_sync")
-	// Start topo server
-	err := clusterInstance.StartTopo()
-	require.NoError(t, err)
-	err = clusterInstance.TopoProcess.ManageTopoDir("mkdir", "/vitess/"+"zone1")
-	require.NoError(t, err)
-	for i := 0; i < 4; i++ {
-		tablet := clusterInstance.NewVttabletInstance("replica", 100+i, "zone1")
-		tablets = append(tablets, tablet)
-	}
-
-	shard := &cluster.Shard{Name: utils.ShardName}
-	shard.Vttablets = tablets
-	clusterInstance.VtTabletExtraArgs = []string{
-		"--lock_tables_timeout", "5s",
-		"--enable_semi_sync",
-		"--init_populate_metadata",
-		"--track_schema_versions=true",
-	}
-
-	// Initialize Cluster
-	err = clusterInstance.SetupCluster(keyspace, []cluster.Shard{*shard})
-	require.NoError(t, err)
-
-	//Start MySql
-	var mysqlCtlProcessList []*exec.Cmd
-	for _, shard := range clusterInstance.Keyspaces[0].Shards {
-		for _, tablet := range shard.Vttablets {
-			log.Infof("Starting MySql for tablet %v", tablet.Alias)
-			proc, err := tablet.MysqlctlProcess.StartProcess()
-			require.NoError(t, err)
-			mysqlCtlProcessList = append(mysqlCtlProcessList, proc)
-		}
-	}
-	// Wait for mysql processes to start
-	for _, proc := range mysqlCtlProcessList {
-		if err := proc.Wait(); err != nil {
-			t.Fatalf("Error starting mysql: %s", err.Error())
-		}
-	}
-
-	for _, tablet := range tablets {
-		// Start the tablet
-		err = tablet.VttabletProcess.Setup()
-		require.NoError(t, err)
-	}
-	for _, tablet := range tablets {
-		err := tablet.VttabletProcess.WaitForTabletStatuses([]string{"SERVING", "NOT_SERVING"})
-		require.NoError(t, err)
-	}
-
-	// Force the replica to reparent assuming that all the datasets are identical.
-	res, err := utils.Ers(clusterInstance, tablets[0], "60s", "30s")
-	require.NoError(t, err, res)
-
-	utils.ValidateTopology(t, clusterInstance, true)
-	// create Tables
-	utils.RunSQL(context.Background(), t, "create table vt_insert_test (id bigint, msg varchar(64), primary key (id)) Engine=InnoDB", tablets[0])
-	utils.CheckPrimaryTablet(t, clusterInstance, tablets[0])
-	utils.ValidateTopology(t, clusterInstance, false)
-	time.Sleep(100 * time.Millisecond) // wait for replication to catchup
-	strArray := utils.GetShardReplicationPositions(t, clusterInstance, utils.KeyspaceName, utils.ShardName, true)
-	assert.Equal(t, len(tablets), len(strArray))
-	assert.Contains(t, strArray[0], "primary") // primary first
-	utils.ConfirmReplication(t, tablets[0], tablets[1:])
-}
-
 func TestRecoverWithMultipleFailures(t *testing.T) {
 	defer cluster.PanicHandler(t)
 	clusterInstance := utils.SetupReparentCluster(t, true)
diff --git a/go/test/endtoend/reparent/utils/utils.go b/go/test/endtoend/reparent/utils/utils.go
index aa559d53a7..ed6b823b61 100644
--- a/go/test/endtoend/reparent/utils/utils.go
+++ b/go/test/endtoend/reparent/utils/utils.go
@@ -90,9 +90,7 @@ func setupCluster(ctx context.Context, t *testing.T, shardName string, cells []s
 
 	if enableSemiSync {
 		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--enable_semi_sync")
-		if clusterInstance.VtctlMajorVersion >= 13 {
-			clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "--durability_policy=semi_sync")
-		}
+		clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "--durability_policy=semi_sync")
 	}
 
 	// Start topo server
@@ -132,17 +130,13 @@ func setupCluster(ctx context.Context, t *testing.T, shardName string, cells []s
 		// In this case, the close method and initSchema method of the onlineDDL executor race.
 		// If the initSchema acquires the lock, then it takes about 30 seconds for it to run during which time the
 		// DemotePrimary rpc is stalled!
-		"--queryserver_enable_online_ddl=false")
-
-	if clusterInstance.VtTabletMajorVersion >= 13 && clusterInstance.VtctlMajorVersion >= 13 {
+		"--queryserver_enable_online_ddl=false",
 		// disabling active reparents on the tablet since we don't want the replication manager
 		// to fix replication if it is stopped. Some tests deliberately do that. Also, we don't want
 		// the replication manager to silently fix the replication in case ERS or PRS mess up. All the
 		// tests in this test suite should work irrespective of this flag. Each run of ERS, PRS should be
 		// setting up the replication correctly.
-		// However, due to the bugs in old vitess components we can only do this for version >= 13.
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--disable_active_reparents")
-	}
+		"--disable_active_reparents")
 
 	// Initialize Cluster
 	err = clusterInstance.SetupCluster(keyspace, []cluster.Shard{*shard})
@@ -208,9 +202,7 @@ func setupClusterLegacy(ctx context.Context, t *testing.T, shardName string, cel
 
 	if enableSemiSync {
 		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--enable_semi_sync")
-		if clusterInstance.VtctlMajorVersion >= 13 {
-			clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "--durability_policy=semi_sync")
-		}
+		clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "--durability_policy=semi_sync")
 	}
 
 	// Start topo server
@@ -250,17 +242,13 @@ func setupClusterLegacy(ctx context.Context, t *testing.T, shardName string, cel
 		// In this case, the close method and initSchema method of the onlineDDL executor race.
 		// If the initSchema acquires the lock, then it takes about 30 seconds for it to run during which time the
 		// DemotePrimary rpc is stalled!
-		"--queryserver_enable_online_ddl=false")
-
-	if clusterInstance.VtTabletMajorVersion >= 13 && clusterInstance.VtctlMajorVersion >= 13 {
+		"--queryserver_enable_online_ddl=false",
 		// disabling active reparents on the tablet since we don't want the replication manager
 		// to fix replication if it is stopped. Some tests deliberately do that. Also, we don't want
 		// the replication manager to silently fix the replication in case ERS or PRS mess up. All the
 		// tests in this test suite should work irrespective of this flag. Each run of ERS, PRS should be
 		// setting up the replication correctly.
-		// However, due to the bugs in old vitess components we can only do this for version >= 13.
-		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--disable_active_reparents")
-	}
+		"--disable_active_reparents")
 
 	// Initialize Cluster
 	err = clusterInstance.SetupCluster(keyspace, []cluster.Shard{*shard})
@@ -423,10 +411,7 @@ func ErsIgnoreTablet(clusterInstance *cluster.LocalProcessCluster, tab *cluster.
 
 // ErsWithVtctl runs ERS via vtctl binary
 func ErsWithVtctl(clusterInstance *cluster.LocalProcessCluster) (string, error) {
-	args := []string{"EmergencyReparentShard", "--", "--keyspace_shard", fmt.Sprintf("%s/%s", KeyspaceName, ShardName)}
-	if clusterInstance.VtctlMajorVersion >= 13 {
-		args = append([]string{"--durability_policy=semi_sync"}, args...)
-	}
+	args := []string{"--durability_policy=semi_sync", "EmergencyReparentShard", "--", "--keyspace_shard", fmt.Sprintf("%s/%s", KeyspaceName, ShardName)}
 	return clusterInstance.VtctlProcess.ExecuteCommandWithOutput(args...)
 }
 
diff --git a/go/test/endtoend/vtorc/general/main_test.go b/go/test/endtoend/vtorc/general/main_test.go
index 604d0e77ec..018e6da21f 100644
--- a/go/test/endtoend/vtorc/general/main_test.go
+++ b/go/test/endtoend/vtorc/general/main_test.go
@@ -52,8 +52,8 @@ func TestMain(m *testing.M) {
 	if clusterInfo != nil {
 		// stop vtorc first otherwise its logs get polluted
 		// with instances being unreachable triggering unnecessary operations
-		if clusterInfo.ClusterInstance.VtorcProcess != nil {
-			_ = clusterInfo.ClusterInstance.VtorcProcess.TearDown()
+		for _, vtorcProcess := range clusterInfo.ClusterInstance.VtorcProcesses {
+			_ = vtorcProcess.TearDown()
 		}
 
 		for _, cellInfo := range clusterInfo.CellInfos {
diff --git a/go/test/endtoend/vtorc/general/vtorc_test.go b/go/test/endtoend/vtorc/general/vtorc_test.go
index 1d1f258bc7..5a413aaa33 100644
--- a/go/test/endtoend/vtorc/general/vtorc_test.go
+++ b/go/test/endtoend/vtorc/general/vtorc_test.go
@@ -32,17 +32,31 @@ import (
 )
 
 // Cases to test:
-// 1. create cluster with 1 replica and 1 rdonly, let orc choose primary
+// 1. create cluster with 2 replicas and 1 rdonly, let orc choose primary
 // verify rdonly is not elected, only replica
 // verify replication is setup
+// verify that with multiple vtorc instances, we still only have 1 PlannedReparentShard call
 func TestPrimaryElection(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 2)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
-	utils.CheckPrimaryTablet(t, clusterInfo, shard0.Vttablets[0], true)
-	utils.CheckReplication(t, clusterInfo, shard0.Vttablets[0], shard0.Vttablets[1:], 10*time.Second)
+	primary := utils.ShardPrimaryTablet(t, clusterInfo, keyspace, shard0)
+	assert.NotNil(t, primary, "should have elected a primary")
+	utils.CheckReplication(t, clusterInfo, primary, shard0.Vttablets, 10*time.Second)
+
+	for _, vttablet := range shard0.Vttablets {
+		if vttablet.Type == "rdonly" && primary.Alias == vttablet.Alias {
+			t.Errorf("Rdonly tablet promoted as primary - %v", primary.Alias)
+		}
+	}
+
+	res, err := utils.RunSQL(t, "select * from reparent_journal", primary, "_vt")
+	require.NoError(t, err)
+	require.Len(t, res.Rows, 1, "There should only be 1 primary tablet which was elected")
 }
 
 // Cases to test:
@@ -51,7 +65,9 @@ func TestPrimaryElection(t *testing.T) {
 // verify replication is setup
 func TestSingleKeyspace(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, []string{"--clusters_to_watch", "ks"}, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, []string{"--clusters_to_watch", "ks"}, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -65,7 +81,9 @@ func TestSingleKeyspace(t *testing.T) {
 // verify replication is setup
 func TestKeyspaceShard(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, []string{"--clusters_to_watch", "ks/0"}, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, []string{"--clusters_to_watch", "ks/0"}, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -95,7 +113,9 @@ func waitForReadOnlyValue(t *testing.T, curPrimary *cluster.Vttablet, expectValu
 // 3. make primary readonly, let orc repair
 func TestPrimaryReadOnly(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -115,7 +135,9 @@ func TestPrimaryReadOnly(t *testing.T) {
 // 4. make replica ReadWrite, let orc repair
 func TestReplicaReadWrite(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -143,7 +165,9 @@ func TestReplicaReadWrite(t *testing.T) {
 // 5. stop replication, let orc repair
 func TestStopReplication(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -175,7 +199,9 @@ func TestStopReplication(t *testing.T) {
 // 6. setup replication from non-primary, let orc repair
 func TestReplicationFromOtherReplica(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 3, 0, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 3, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -221,7 +247,9 @@ func TestRepairAfterTER(t *testing.T) {
 	// test fails intermittently on CI, skip until it can be fixed.
 	t.SkipNow()
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -252,7 +280,9 @@ func TestRepairAfterTER(t *testing.T) {
 // 7. make instance A replicates from B and B from A, wait for repair
 func TestCircularReplication(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -292,11 +322,14 @@ func TestCircularReplication(t *testing.T) {
 // TestSemiSync tests that semi-sync is setup correctly by vtorc if it is incorrectly set
 func TestSemiSync(t *testing.T) {
 	// stop any vtorc instance running due to a previous test.
-	utils.StopVtorc(t, clusterInfo)
+	utils.StopVtorcs(t, clusterInfo)
 	newCluster := utils.SetupNewClusterSemiSync(t)
-	utils.StartVtorc(t, newCluster, nil, "test_config_semi_sync.json")
+	utils.StartVtorcs(t, newCluster, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+		Durability:                            "semi_sync",
+	}, 1)
 	defer func() {
-		utils.StopVtorc(t, newCluster)
+		utils.StopVtorcs(t, newCluster)
 		newCluster.ClusterInstance.Teardown()
 	}()
 	keyspace := &newCluster.ClusterInstance.Keyspaces[0]
@@ -353,3 +386,43 @@ func TestSemiSync(t *testing.T) {
 		}
 	}
 }
+
+// TestVtorcWithPrs tests that VTOrc works fine even when PRS is called from vtctld
+func TestVtorcWithPrs(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 4, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
+	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
+	shard0 := &keyspace.Shards[0]
+
+	// find primary from topo
+	curPrimary := utils.ShardPrimaryTablet(t, clusterInfo, keyspace, shard0)
+	assert.NotNil(t, curPrimary, "should have elected a primary")
+
+	// find any replica tablet other than the current primary
+	var replica *cluster.Vttablet
+	for _, tablet := range shard0.Vttablets {
+		if tablet.Alias != curPrimary.Alias {
+			replica = tablet
+			break
+		}
+	}
+	assert.NotNil(t, replica, "could not find any replica tablet")
+
+	// check that the replication is setup correctly before we failover
+	utils.CheckReplication(t, clusterInfo, curPrimary, shard0.Vttablets, 10*time.Second)
+
+	output, err := clusterInfo.ClusterInstance.VtctlclientProcess.ExecuteCommandWithOutput(
+		"PlannedReparentShard",
+		"-keyspace_shard", fmt.Sprintf("%s/%s", keyspace.Name, shard0.Name),
+		"-wait_replicas_timeout", "31s",
+		"-new_primary", replica.Alias)
+	require.NoError(t, err, "error in PlannedReparentShard output - %s", output)
+
+	time.Sleep(40 * time.Second)
+
+	// check that the replica gets promoted
+	utils.CheckPrimaryTablet(t, clusterInfo, replica, true)
+	utils.VerifyWritesSucceed(t, clusterInfo, replica, shard0.Vttablets, 10*time.Second)
+}
diff --git a/go/test/endtoend/vtorc/gracefultakeover/graceful_takeover_test.go b/go/test/endtoend/vtorc/gracefultakeover/graceful_takeover_test.go
index 41b3a21bdf..ddb6649d59 100644
--- a/go/test/endtoend/vtorc/gracefultakeover/graceful_takeover_test.go
+++ b/go/test/endtoend/vtorc/gracefultakeover/graceful_takeover_test.go
@@ -31,7 +31,9 @@ import (
 // covers the test case graceful-master-takeover from orchestrator
 func TestGracefulPrimaryTakeover(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -52,7 +54,7 @@ func TestGracefulPrimaryTakeover(t *testing.T) {
 	// check that the replication is setup correctly before we failover
 	utils.CheckReplication(t, clusterInfo, curPrimary, []*cluster.Vttablet{replica}, 10*time.Second)
 
-	status, _ := utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:3000/api/graceful-primary-takeover/localhost/%d/localhost/%d", curPrimary.MySQLPort, replica.MySQLPort))
+	status, _ := utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:%d/api/graceful-primary-takeover/localhost/%d/localhost/%d", clusterInfo.ClusterInstance.VtorcProcesses[0].WebPort, curPrimary.MySQLPort, replica.MySQLPort))
 	assert.Equal(t, 200, status)
 
 	// check that the replica gets promoted
@@ -65,7 +67,9 @@ func TestGracefulPrimaryTakeover(t *testing.T) {
 // orchestrator used to fail in this case, but for VtOrc, specifying no target makes it choose one on its own
 func TestGracefulPrimaryTakeoverNoTarget(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 0, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -86,7 +90,7 @@ func TestGracefulPrimaryTakeoverNoTarget(t *testing.T) {
 	// check that the replication is setup correctly before we failover
 	utils.CheckReplication(t, clusterInfo, curPrimary, []*cluster.Vttablet{replica}, 10*time.Second)
 
-	status, _ := utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:3000/api/graceful-primary-takeover/localhost/%d/", curPrimary.MySQLPort))
+	status, _ := utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:%d/api/graceful-primary-takeover/localhost/%d/", clusterInfo.ClusterInstance.VtorcProcesses[0].WebPort, curPrimary.MySQLPort))
 	assert.Equal(t, 200, status)
 
 	// check that the replica gets promoted
@@ -98,7 +102,9 @@ func TestGracefulPrimaryTakeoverNoTarget(t *testing.T) {
 // covers the test case graceful-master-takeover-auto from orchestrator
 func TestGracefulPrimaryTakeoverAuto(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -123,14 +129,14 @@ func TestGracefulPrimaryTakeoverAuto(t *testing.T) {
 	// check that the replication is setup correctly before we failover
 	utils.CheckReplication(t, clusterInfo, primary, []*cluster.Vttablet{replica, rdonly}, 10*time.Second)
 
-	status, _ := utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:3000/api/graceful-primary-takeover-auto/localhost/%d/localhost/%d", primary.MySQLPort, replica.MySQLPort))
+	status, _ := utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:%d/api/graceful-primary-takeover-auto/localhost/%d/localhost/%d", clusterInfo.ClusterInstance.VtorcProcesses[0].WebPort, primary.MySQLPort, replica.MySQLPort))
 	assert.Equal(t, 200, status)
 
 	// check that the replica gets promoted
 	utils.CheckPrimaryTablet(t, clusterInfo, replica, true)
 	utils.VerifyWritesSucceed(t, clusterInfo, replica, []*cluster.Vttablet{primary, rdonly}, 10*time.Second)
 
-	status, _ = utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:3000/api/graceful-primary-takeover-auto/localhost/%d/", replica.MySQLPort))
+	status, _ = utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:%d/api/graceful-primary-takeover-auto/localhost/%d/", clusterInfo.ClusterInstance.VtorcProcesses[0].WebPort, replica.MySQLPort))
 	assert.Equal(t, 200, status)
 
 	// check that the primary gets promoted back
@@ -142,7 +148,9 @@ func TestGracefulPrimaryTakeoverAuto(t *testing.T) {
 // covers the test case graceful-master-takeover-fail-cross-region from orchestrator
 func TestGracefulPrimaryTakeoverFailCrossCell(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 
@@ -163,7 +171,7 @@ func TestGracefulPrimaryTakeoverFailCrossCell(t *testing.T) {
 	// newly started tablet does not replicate from anyone yet, we will allow orchestrator to fix this too
 	utils.CheckReplication(t, clusterInfo, primary, []*cluster.Vttablet{crossCellReplica1, rdonly}, 25*time.Second)
 
-	status, response := utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:3000/api/graceful-primary-takeover/localhost/%d/localhost/%d", primary.MySQLPort, crossCellReplica1.MySQLPort))
+	status, response := utils.MakeAPICallUntilRegistered(t, fmt.Sprintf("http://localhost:%d/api/graceful-primary-takeover/localhost/%d/localhost/%d", clusterInfo.ClusterInstance.VtorcProcesses[0].WebPort, primary.MySQLPort, crossCellReplica1.MySQLPort))
 	assert.Equal(t, 500, status)
 	assert.Contains(t, response, "GracefulPrimaryTakeover: constraint failure")
 
diff --git a/go/test/endtoend/vtorc/gracefultakeover/main_test.go b/go/test/endtoend/vtorc/gracefultakeover/main_test.go
index 8af51b0fc3..d429e36a35 100644
--- a/go/test/endtoend/vtorc/gracefultakeover/main_test.go
+++ b/go/test/endtoend/vtorc/gracefultakeover/main_test.go
@@ -58,8 +58,8 @@ func TestMain(m *testing.M) {
 	if clusterInfo != nil {
 		// stop vtorc first otherwise its logs get polluted
 		// with instances being unreachable triggering unnecessary operations
-		if clusterInfo.ClusterInstance.VtorcProcess != nil {
-			_ = clusterInfo.ClusterInstance.VtorcProcess.TearDown()
+		for _, vtorcProcess := range clusterInfo.ClusterInstance.VtorcProcesses {
+			_ = vtorcProcess.TearDown()
 		}
 
 		for _, cellInfo := range clusterInfo.CellInfos {
diff --git a/go/test/endtoend/vtorc/primaryfailure/main_test.go b/go/test/endtoend/vtorc/primaryfailure/main_test.go
index 6761d4b188..73a159ceda 100644
--- a/go/test/endtoend/vtorc/primaryfailure/main_test.go
+++ b/go/test/endtoend/vtorc/primaryfailure/main_test.go
@@ -59,8 +59,8 @@ func TestMain(m *testing.M) {
 	if clusterInfo != nil {
 		// stop vtorc first otherwise its logs get polluted
 		// with instances being unreachable triggering unnecessary operations
-		if clusterInfo.ClusterInstance.VtorcProcess != nil {
-			_ = clusterInfo.ClusterInstance.VtorcProcess.TearDown()
+		for _, vtorcProcess := range clusterInfo.ClusterInstance.VtorcProcesses {
+			_ = vtorcProcess.TearDown()
 		}
 
 		for _, cellInfo := range clusterInfo.CellInfos {
diff --git a/go/test/endtoend/vtorc/primaryfailure/primary_failure_test.go b/go/test/endtoend/vtorc/primaryfailure/primary_failure_test.go
index 85ea91eed2..a56a27d69f 100644
--- a/go/test/endtoend/vtorc/primaryfailure/primary_failure_test.go
+++ b/go/test/endtoend/vtorc/primaryfailure/primary_failure_test.go
@@ -32,7 +32,9 @@ import (
 // covers the test case master-failover from orchestrator
 func TestDownPrimary(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
@@ -74,7 +76,9 @@ func TestDownPrimary(t *testing.T) {
 // covers part of the test case master-failover-lost-replicas from orchestrator
 func TestCrossDataCenterFailure(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
@@ -117,7 +121,9 @@ func TestCrossDataCenterFailure(t *testing.T) {
 // In case of no viable candidates, we should error out
 func TestCrossDataCenterFailureError(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 1, 1, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
@@ -161,7 +167,9 @@ func TestLostRdonlyOnPrimaryFailure(t *testing.T) {
 	// were detected by vtorc and could be configured to have their sources detached
 	t.Skip()
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 2, nil, "test_config.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 2, nil, cluster.VtorcConfiguration{
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
@@ -240,7 +248,10 @@ func TestLostRdonlyOnPrimaryFailure(t *testing.T) {
 // covers the test case master-failover-fail-promotion-lag-minutes-success from orchestrator
 func TestPromotionLagSuccess(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, "test_config_promotion_success.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, cluster.VtorcConfiguration{
+		ReplicationLagQuery:              "select 59",
+		FailPrimaryPromotionOnLagMinutes: 1,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
@@ -286,7 +297,10 @@ func TestPromotionLagFailure(t *testing.T) {
 	// was smaller than the configured value, otherwise it would fail the promotion
 	t.Skip()
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 3, 1, nil, "test_config_promotion_failure.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 3, 1, nil, cluster.VtorcConfiguration{
+		ReplicationLagQuery:              "select 61",
+		FailPrimaryPromotionOnLagMinutes: 1,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
@@ -335,7 +349,10 @@ func TestPromotionLagFailure(t *testing.T) {
 // That is the replica which should be promoted in case of primary failure
 func TestDownPrimaryPromotionRule(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, "test_config_crosscenter_prefer.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, cluster.VtorcConfiguration{
+		LockShardTimeoutSeconds: 5,
+		Durability:              "test",
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
@@ -380,7 +397,10 @@ func TestDownPrimaryPromotionRule(t *testing.T) {
 // It should also be caught up when it is promoted
 func TestDownPrimaryPromotionRuleWithLag(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, "test_config_crosscenter_prefer.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, cluster.VtorcConfiguration{
+		LockShardTimeoutSeconds: 5,
+		Durability:              "test",
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
@@ -457,7 +477,11 @@ func TestDownPrimaryPromotionRuleWithLag(t *testing.T) {
 // It should also be caught up when it is promoted
 func TestDownPrimaryPromotionRuleWithLagCrossCenter(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, "test_config_crosscenter_prefer_prevent.json")
+	utils.SetupVttabletsAndVtorc(t, clusterInfo, 2, 1, nil, cluster.VtorcConfiguration{
+		LockShardTimeoutSeconds:               5,
+		Durability:                            "test",
+		PreventCrossDataCenterPrimaryFailover: true,
+	}, 1)
 	keyspace := &clusterInfo.ClusterInstance.Keyspaces[0]
 	shard0 := &keyspace.Shards[0]
 	// find primary from topo
diff --git a/go/test/endtoend/vtorc/utils/test_config.json b/go/test/endtoend/vtorc/utils/test_config.json
deleted file mode 100644
index 8ba2885009..0000000000
--- a/go/test/endtoend/vtorc/utils/test_config.json
+++ /dev/null
@@ -1,10 +0,0 @@
-{
-  "Debug": true,
-  "MySQLTopologyUser": "orc_client_user",
-  "MySQLTopologyPassword": "orc_client_user_password",
-  "MySQLReplicaUser": "vt_repl",
-  "MySQLReplicaPassword": "",
-  "RecoveryPeriodBlockSeconds": 1,
-  "InstancePollSeconds": 1,
-  "PreventCrossDataCenterPrimaryFailover": true
-}
diff --git a/go/test/endtoend/vtorc/utils/test_config_crosscenter_prefer.json b/go/test/endtoend/vtorc/utils/test_config_crosscenter_prefer.json
deleted file mode 100644
index c9d0974736..0000000000
--- a/go/test/endtoend/vtorc/utils/test_config_crosscenter_prefer.json
+++ /dev/null
@@ -1,11 +0,0 @@
-{
-  "Debug": true,
-  "MySQLTopologyUser": "orc_client_user",
-  "MySQLTopologyPassword": "orc_client_user_password",
-  "MySQLReplicaUser": "vt_repl",
-  "MySQLReplicaPassword": "",
-  "RecoveryPeriodBlockSeconds": 1,
-  "InstancePollSeconds": 1,
-  "LockShardTimeoutSeconds": 5,
-  "Durability": "test"
-}
diff --git a/go/test/endtoend/vtorc/utils/test_config_crosscenter_prefer_prevent.json b/go/test/endtoend/vtorc/utils/test_config_crosscenter_prefer_prevent.json
deleted file mode 100644
index cb4ef5000f..0000000000
--- a/go/test/endtoend/vtorc/utils/test_config_crosscenter_prefer_prevent.json
+++ /dev/null
@@ -1,12 +0,0 @@
-{
-  "Debug": true,
-  "MySQLTopologyUser": "orc_client_user",
-  "MySQLTopologyPassword": "orc_client_user_password",
-  "MySQLReplicaUser": "vt_repl",
-  "MySQLReplicaPassword": "",
-  "RecoveryPeriodBlockSeconds": 1,
-  "InstancePollSeconds": 1,
-  "LockShardTimeoutSeconds": 5,
-  "Durability": "test",
-  "PreventCrossDataCenterPrimaryFailover": true
-}
diff --git a/go/test/endtoend/vtorc/utils/test_config_promotion_failure.json b/go/test/endtoend/vtorc/utils/test_config_promotion_failure.json
deleted file mode 100644
index 94daa6a729..0000000000
--- a/go/test/endtoend/vtorc/utils/test_config_promotion_failure.json
+++ /dev/null
@@ -1,11 +0,0 @@
-{
-  "Debug": true,
-  "MySQLTopologyUser": "orc_client_user",
-  "MySQLTopologyPassword": "orc_client_user_password",
-  "MySQLReplicaUser": "vt_repl",
-  "MySQLReplicaPassword": "",
-  "RecoveryPeriodBlockSeconds": 1,
-  "InstancePollSeconds": 1,
-  "ReplicationLagQuery": "select 61",
-  "FailPrimaryPromotionOnLagMinutes": 1
-}
diff --git a/go/test/endtoend/vtorc/utils/test_config_promotion_success.json b/go/test/endtoend/vtorc/utils/test_config_promotion_success.json
deleted file mode 100644
index 5084e4598e..0000000000
--- a/go/test/endtoend/vtorc/utils/test_config_promotion_success.json
+++ /dev/null
@@ -1,11 +0,0 @@
-{
-  "Debug": true,
-  "MySQLTopologyUser": "orc_client_user",
-  "MySQLTopologyPassword": "orc_client_user_password",
-  "MySQLReplicaUser": "vt_repl",
-  "MySQLReplicaPassword": "",
-  "RecoveryPeriodBlockSeconds": 1,
-  "InstancePollSeconds": 1,
-  "ReplicationLagQuery": "select 59",
-  "FailPrimaryPromotionOnLagMinutes": 1
-}
diff --git a/go/test/endtoend/vtorc/utils/test_config_semi_sync.json b/go/test/endtoend/vtorc/utils/test_config_semi_sync.json
deleted file mode 100644
index 8b531fdc6a..0000000000
--- a/go/test/endtoend/vtorc/utils/test_config_semi_sync.json
+++ /dev/null
@@ -1,11 +0,0 @@
-{
-  "Debug": true,
-  "MySQLTopologyUser": "orc_client_user",
-  "MySQLTopologyPassword": "orc_client_user_password",
-  "MySQLReplicaUser": "vt_repl",
-  "MySQLReplicaPassword": "",
-  "Durability": "semi_sync",
-  "RecoveryPeriodBlockSeconds": 1,
-  "InstancePollSeconds": 1,
-  "PreventCrossDataCenterPrimaryFailover": true
-}
diff --git a/go/test/endtoend/vtorc/utils/utils.go b/go/test/endtoend/vtorc/utils/utils.go
index 1e2d308a81..4641a898be 100644
--- a/go/test/endtoend/vtorc/utils/utils.go
+++ b/go/test/endtoend/vtorc/utils/utils.go
@@ -240,38 +240,35 @@ func demotePrimaryTablet(ts *topo.Server) (err error) {
 	return
 }
 
-// StartVtorc is used to start the orchestrator with the given extra arguments
-func StartVtorc(t *testing.T, clusterInfo *VtOrcClusterInfo, orcExtraArgs []string, configFileName string) {
+// StartVtorcs is used to start the orchestrator with the given extra arguments
+func StartVtorcs(t *testing.T, clusterInfo *VtOrcClusterInfo, orcExtraArgs []string, config cluster.VtorcConfiguration, count int) {
 	t.Helper()
-	workingDir := os.Getenv("PWD")
-	idx := strings.Index(workingDir, "vtorc")
-	if idx == -1 {
-		require.Fail(t, "SetupVttabletsAndVtorc should only be used from a package inside the vtorc directory")
-	}
-
-	pathToConfig := path.Join(workingDir[:idx], "vtorc", "utils", configFileName)
 	// Start vtorc
-	clusterInfo.ClusterInstance.VtorcProcess = clusterInfo.ClusterInstance.NewOrcProcess(pathToConfig)
-	clusterInfo.ClusterInstance.VtorcProcess.ExtraArgs = orcExtraArgs
-	err := clusterInfo.ClusterInstance.VtorcProcess.Setup()
-	require.NoError(t, err)
+	for i := 0; i < count; i++ {
+		vtorcProcess := clusterInfo.ClusterInstance.NewOrcProcess(config)
+		vtorcProcess.ExtraArgs = orcExtraArgs
+		err := vtorcProcess.Setup()
+		require.NoError(t, err)
+		clusterInfo.ClusterInstance.VtorcProcesses = append(clusterInfo.ClusterInstance.VtorcProcesses, vtorcProcess)
+	}
 }
 
-// StopVtorc is used to stop the orchestrator
-func StopVtorc(t *testing.T, clusterInfo *VtOrcClusterInfo) {
+// StopVtorcs is used to stop the orchestrator
+func StopVtorcs(t *testing.T, clusterInfo *VtOrcClusterInfo) {
 	t.Helper()
 	// Stop vtorc
-	if clusterInfo.ClusterInstance.VtorcProcess != nil {
-		err := clusterInfo.ClusterInstance.VtorcProcess.TearDown()
-		require.NoError(t, err)
+	for _, vtorcProcess := range clusterInfo.ClusterInstance.VtorcProcesses {
+		if err := vtorcProcess.TearDown(); err != nil {
+			log.Errorf("Error in vtorc teardown: %v", err)
+		}
 	}
-	clusterInfo.ClusterInstance.VtorcProcess = nil
+	clusterInfo.ClusterInstance.VtorcProcesses = nil
 }
 
 // SetupVttabletsAndVtorc is used to setup the vttablets and start the orchestrator
-func SetupVttabletsAndVtorc(t *testing.T, clusterInfo *VtOrcClusterInfo, numReplicasReqCell1, numRdonlyReqCell1 int, orcExtraArgs []string, configFileName string) {
+func SetupVttabletsAndVtorc(t *testing.T, clusterInfo *VtOrcClusterInfo, numReplicasReqCell1, numRdonlyReqCell1 int, orcExtraArgs []string, config cluster.VtorcConfiguration, vtorcCount int) {
 	// stop vtorc if it is running
-	StopVtorc(t, clusterInfo)
+	StopVtorcs(t, clusterInfo)
 
 	// remove all the vttablets so that each test can add the amount that they require
 	err := shutdownVttablets(clusterInfo)
@@ -312,7 +309,7 @@ func SetupVttabletsAndVtorc(t *testing.T, clusterInfo *VtOrcClusterInfo, numRepl
 	}
 
 	// start vtorc
-	StartVtorc(t, clusterInfo, orcExtraArgs, configFileName)
+	StartVtorcs(t, clusterInfo, orcExtraArgs, config, vtorcCount)
 }
 
 // cleanAndStartVttablet cleans the MySQL instance underneath for running a new test. It also starts the vttablet.
diff --git a/go/vt/vtctl/reparentutil/planned_reparenter_flaky_test.go b/go/vt/vtctl/reparentutil/planned_reparenter_flaky_test.go
index 2c4091ffe4..51ee27bb30 100644
--- a/go/vt/vtctl/reparentutil/planned_reparenter_flaky_test.go
+++ b/go/vt/vtctl/reparentutil/planned_reparenter_flaky_test.go
@@ -278,15 +278,44 @@ func TestPlannedReparenter_ReparentShard(t *testing.T) {
 			},
 		},
 		{
-			name: "cannot lock shard",
+			name: "already locked shard",
 			ts:   memorytopo.NewServer("zone1"),
-			tmc:  nil,
+			tmc: &testutil.TabletManagerClient{
+				PrimaryPositionResults: map[string]struct {
+					Position string
+					Error    error
+				}{
+					"zone1-0000000100": {
+						Position: "position1",
+						Error:    nil,
+					},
+				},
+				PopulateReparentJournalResults: map[string]error{
+					"zone1-0000000100": nil,
+				},
+				SetReplicationSourceResults: map[string]error{
+					"zone1-0000000200": nil,
+				},
+				SetReadWriteResults: map[string]error{
+					"zone1-0000000100": nil,
+				},
+			},
 			tablets: []*topodatapb.Tablet{
 				{
 					Alias: &topodatapb.TabletAlias{
 						Cell: "zone1",
 						Uid:  100,
 					},
+					Type:     topodatapb.TabletType_PRIMARY,
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  200,
+					},
+					Type:     topodatapb.TabletType_REPLICA,
 					Keyspace: "testkeyspace",
 					Shard:    "-",
 				},
@@ -295,10 +324,33 @@ func TestPlannedReparenter_ReparentShard(t *testing.T) {
 
 			keyspace: "testkeyspace",
 			shard:    "-",
-			opts:     PlannedReparentOptions{},
+			opts: PlannedReparentOptions{
+				NewPrimaryAlias: &topodatapb.TabletAlias{
+					Cell: "zone1",
+					Uid:  100,
+				},
+			},
 
-			expectedEvent: nil,
-			shouldErr:     true,
+			shouldErr: false,
+			expectedEvent: &events.Reparent{
+				ShardInfo: *topo.NewShardInfo("testkeyspace", "-", &topodatapb.Shard{
+					PrimaryAlias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					KeyRange:         &topodatapb.KeyRange{},
+					IsPrimaryServing: true,
+				}, nil),
+				NewPrimary: &topodatapb.Tablet{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Type:     topodatapb.TabletType_PRIMARY,
+					Keyspace: "testkeyspace",
+					Shard:    "-",
+				},
+			},
 		},
 		{
 			// The simplest setup required to make an overall ReparentShard call
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index 0e342f6d61..3810378a82 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
@@ -3895,3 +3895,55 @@ Gen4 plan same as above
 "select count(distinct user_id, name) from user"
 "unsupported: only one expression allowed inside aggregates: count(distinct user_id, `name`)"
 Gen4 error: aggregate functions take a single argument 'count(distinct user_id, `name`)'
+
+"select sum(col) from (select user.col as col, 32 from user join user_extra) t"
+"unsupported: cross-shard query with aggregates"
+{
+  "QueryType": "SELECT",
+  "Original": "select sum(col) from (select user.col as col, 32 from user join user_extra) t",
+  "Instructions": {
+    "OperatorType": "Aggregate",
+    "Variant": "Scalar",
+    "Aggregates": "sum(0) AS sum(col)",
+    "Inputs": [
+      {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "[COLUMN 2] * [COLUMN 3] as sum(col)"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Join",
+            "Variant": "Join",
+            "JoinColumnIndexes": "L:0,L:1,L:2,R:0",
+            "TableName": "`user`_user_extra",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select `user`.col as col, 32, sum(col) from `user` where 1 != 1",
+                "Query": "select `user`.col as col, 32, sum(col) from `user`",
+                "Table": "`user`"
+              },
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select count(*) from user_extra where 1 != 1",
+                "Query": "select count(*) from user_extra",
+                "Table": "user_extra"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt b/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
index e8ec3383e2..74b962f20a 100644
--- a/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/tpch_cases.txt
@@ -418,7 +418,182 @@ Gen4 error: unsupported: cross-shard correlated subquery
 # TPC-H query 7
 "select supp_nation, cust_nation, l_year, sum(volume) as revenue from (select n1.n_name as supp_nation, n2.n_name as cust_nation, extract(year from l_shipdate) as l_year, l_extendedprice * (1 - l_discount) as volume from supplier, lineitem, orders, customer, nation n1, nation n2 where s_suppkey = l_suppkey and o_orderkey = l_orderkey and c_custkey = o_custkey and s_nationkey = n1.n_nationkey and c_nationkey = n2.n_nationkey and ((n1.n_name = 'FRANCE' and n2.n_name = 'GERMANY') or (n1.n_name = 'GERMANY' and n2.n_name = 'FRANCE')) and l_shipdate between date('1995-01-01') and date('1996-12-31')) as shipping group by supp_nation, cust_nation, l_year order by supp_nation, cust_nation, l_year"
 "unsupported: cross-shard query with aggregates"
-Gen4 error: using aggregation on top of a *planbuilder.simpleProjection plan is not yet supported
+{
+  "QueryType": "SELECT",
+  "Original": "select supp_nation, cust_nation, l_year, sum(volume) as revenue from (select n1.n_name as supp_nation, n2.n_name as cust_nation, extract(year from l_shipdate) as l_year, l_extendedprice * (1 - l_discount) as volume from supplier, lineitem, orders, customer, nation n1, nation n2 where s_suppkey = l_suppkey and o_orderkey = l_orderkey and c_custkey = o_custkey and s_nationkey = n1.n_nationkey and c_nationkey = n2.n_nationkey and ((n1.n_name = 'FRANCE' and n2.n_name = 'GERMANY') or (n1.n_name = 'GERMANY' and n2.n_name = 'FRANCE')) and l_shipdate between date('1995-01-01') and date('1996-12-31')) as shipping group by supp_nation, cust_nation, l_year order by supp_nation, cust_nation, l_year",
+  "Instructions": {
+    "OperatorType": "Aggregate",
+    "Variant": "Ordered",
+    "Aggregates": "sum(3) AS revenue",
+    "GroupBy": "(0|6), (1|5), (2|4)",
+    "ResultColumns": 4,
+    "Inputs": [
+      {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "[COLUMN 4]",
+          "[COLUMN 5]",
+          "[COLUMN 6]",
+          "(((([COLUMN 10] * [COLUMN 11]) * [COLUMN 12]) * [COLUMN 13]) * [COLUMN 14]) * [COLUMN 15] as revenue",
+          "[COLUMN 9]",
+          "[COLUMN 8]",
+          "[COLUMN 7]"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Sort",
+            "Variant": "Memory",
+            "OrderBy": "(0|16) ASC, (1|17) ASC, (2|18) ASC",
+            "Inputs": [
+              {
+                "OperatorType": "Join",
+                "Variant": "Join",
+                "JoinColumnIndexes": "L:3,R:0,L:4,L:5,L:9,R:1,L:10,L:14,R:2,L:15,L:16,L:17,L:18,L:19,R:3,R:4,L:20,R:5,L:21",
+                "JoinVars": {
+                  "n1_n_name": 2,
+                  "o_custkey": 0
+                },
+                "TableName": "lineitem_orders_supplier_nation_customer_nation",
+                "Inputs": [
+                  {
+                    "OperatorType": "Join",
+                    "Variant": "Join",
+                    "JoinColumnIndexes": "L:1,R:0,R:1,R:2,L:2,L:3,L:5,R:3,R:4,R:5,L:6,L:8,R:6,R:7,R:8,L:9,L:10,L:11,R:9,R:10,R:11,L:12",
+                    "JoinVars": {
+                      "l_suppkey": 0
+                    },
+                    "TableName": "lineitem_orders_supplier_nation",
+                    "Inputs": [
+                      {
+                        "OperatorType": "Join",
+                        "Variant": "Join",
+                        "JoinColumnIndexes": "L:1,R:0,L:2,L:3,L:1,R:0,L:2,L:6,R:2,L:7,L:4,R:1,L:8",
+                        "JoinVars": {
+                          "l_orderkey": 0
+                        },
+                        "TableName": "lineitem_orders",
+                        "Inputs": [
+                          {
+                            "OperatorType": "Route",
+                            "Variant": "Scatter",
+                            "Keyspace": {
+                              "Name": "main",
+                              "Sharded": true
+                            },
+                            "FieldQuery": "select l_orderkey, l_suppkey, extract(year from l_shipdate) as l_year, l_extendedprice * (1 - l_discount) as volume, sum(volume) as revenue, weight_string(l_orderkey), weight_string(l_suppkey), weight_string(extract(year from l_shipdate)), weight_string(extract(year from l_shipdate)) from lineitem where 1 != 1 group by l_orderkey, weight_string(l_orderkey), l_suppkey, weight_string(l_suppkey), l_year, weight_string(l_year)",
+                            "Query": "select l_orderkey, l_suppkey, extract(year from l_shipdate) as l_year, l_extendedprice * (1 - l_discount) as volume, sum(volume) as revenue, weight_string(l_orderkey), weight_string(l_suppkey), weight_string(extract(year from l_shipdate)), weight_string(extract(year from l_shipdate)) from lineitem where l_shipdate between date('1995-01-01') and date('1996-12-31') group by l_orderkey, weight_string(l_orderkey), l_suppkey, weight_string(l_suppkey), l_year, weight_string(l_year)",
+                            "Table": "lineitem"
+                          },
+                          {
+                            "OperatorType": "Route",
+                            "Variant": "EqualUnique",
+                            "Keyspace": {
+                              "Name": "main",
+                              "Sharded": true
+                            },
+                            "FieldQuery": "select o_custkey, count(*), weight_string(o_custkey) from orders where 1 != 1 group by o_custkey, weight_string(o_custkey)",
+                            "Query": "select o_custkey, count(*), weight_string(o_custkey) from orders where o_orderkey = :l_orderkey group by o_custkey, weight_string(o_custkey)",
+                            "Table": "orders",
+                            "Values": [
+                              ":l_orderkey"
+                            ],
+                            "Vindex": "hash"
+                          }
+                        ]
+                      },
+                      {
+                        "OperatorType": "Join",
+                        "Variant": "Join",
+                        "JoinColumnIndexes": "R:0,R:0,R:1,R:0,R:0,R:1,R:3,R:3,R:4,L:1,R:2,R:5",
+                        "JoinVars": {
+                          "s_nationkey": 0
+                        },
+                        "TableName": "supplier_nation",
+                        "Inputs": [
+                          {
+                            "OperatorType": "Route",
+                            "Variant": "EqualUnique",
+                            "Keyspace": {
+                              "Name": "main",
+                              "Sharded": true
+                            },
+                            "FieldQuery": "select s_nationkey, count(*), weight_string(s_nationkey) from supplier where 1 != 1 group by s_nationkey, weight_string(s_nationkey)",
+                            "Query": "select s_nationkey, count(*), weight_string(s_nationkey) from supplier where s_suppkey = :l_suppkey group by s_nationkey, weight_string(s_nationkey)",
+                            "Table": "supplier",
+                            "Values": [
+                              ":l_suppkey"
+                            ],
+                            "Vindex": "hash"
+                          },
+                          {
+                            "OperatorType": "Route",
+                            "Variant": "EqualUnique",
+                            "Keyspace": {
+                              "Name": "main",
+                              "Sharded": true
+                            },
+                            "FieldQuery": "select n1.n_name, n1.n_name as supp_nation, count(*), weight_string(n1.n_name), weight_string(n1.n_name), weight_string(n1.n_name) from nation as n1 where 1 != 1 group by n1.n_name, weight_string(n1.n_name), supp_nation, weight_string(supp_nation)",
+                            "Query": "select n1.n_name, n1.n_name as supp_nation, count(*), weight_string(n1.n_name), weight_string(n1.n_name), weight_string(n1.n_name) from nation as n1 where n1.n_nationkey = :s_nationkey group by n1.n_name, weight_string(n1.n_name), supp_nation, weight_string(supp_nation)",
+                            "Table": "nation",
+                            "Values": [
+                              ":s_nationkey"
+                            ],
+                            "Vindex": "hash"
+                          }
+                        ]
+                      }
+                    ]
+                  },
+                  {
+                    "OperatorType": "Join",
+                    "Variant": "Join",
+                    "JoinColumnIndexes": "R:0,R:0,R:2,L:1,R:1,R:3",
+                    "JoinVars": {
+                      "c_nationkey": 0
+                    },
+                    "TableName": "customer_nation",
+                    "Inputs": [
+                      {
+                        "OperatorType": "Route",
+                        "Variant": "EqualUnique",
+                        "Keyspace": {
+                          "Name": "main",
+                          "Sharded": true
+                        },
+                        "FieldQuery": "select c_nationkey, count(*), weight_string(c_nationkey) from customer where 1 != 1 group by c_nationkey, weight_string(c_nationkey)",
+                        "Query": "select c_nationkey, count(*), weight_string(c_nationkey) from customer where c_custkey = :o_custkey group by c_nationkey, weight_string(c_nationkey)",
+                        "Table": "customer",
+                        "Values": [
+                          ":o_custkey"
+                        ],
+                        "Vindex": "hash"
+                      },
+                      {
+                        "OperatorType": "Route",
+                        "Variant": "EqualUnique",
+                        "Keyspace": {
+                          "Name": "main",
+                          "Sharded": true
+                        },
+                        "FieldQuery": "select n2.n_name as cust_nation, count(*), weight_string(n2.n_name), weight_string(n2.n_name) from nation as n2 where 1 != 1 group by cust_nation, weight_string(cust_nation)",
+                        "Query": "select n2.n_name as cust_nation, count(*), weight_string(n2.n_name), weight_string(n2.n_name) from nation as n2 where n2.n_nationkey = :c_nationkey and (:n1_n_name = 'FRANCE' and n2.n_name = 'GERMANY' or :n1_n_name = 'GERMANY' and n2.n_name = 'FRANCE') group by cust_nation, weight_string(cust_nation)",
+                        "Table": "nation",
+                        "Values": [
+                          ":c_nationkey"
+                        ],
+                        "Vindex": "hash"
+                      }
+                    ]
+                  }
+                ]
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
 
 # TPC-H query 8
 "select o_year, sum(case when nation = 'BRAZIL' then volume else 0 end) / sum(volume) as mkt_share from ( select extract(year from o_orderdate) as o_year, l_extendedprice * (1 - l_discount) as volume, n2.n_name as nation from part, supplier, lineitem, orders, customer, nation n1, nation n2, region where p_partkey = l_partkey and s_suppkey = l_suppkey and l_orderkey = o_orderkey and o_custkey = c_custkey and c_nationkey = n1.n_nationkey and n1.n_regionkey = r_regionkey and r_name = 'AMERICA' and s_nationkey = n2.n_nationkey and o_orderdate between date '1995-01-01' and date('1996-12-31') and p_type = 'ECONOMY ANODIZED STEEL' ) as all_nations group by o_year order by o_year"
@@ -428,7 +603,7 @@ Gen4 plan same as above
 # TPC-H query 9
 "select nation, o_year, sum(amount) as sum_profit from ( select n_name as nation, extract(year from o_orderdate) as o_year, l_extendedprice * (1 - l_discount) - ps_supplycost * l_quantity as amount from part, supplier, lineitem, partsupp, orders, nation where s_suppkey = l_suppkey and ps_suppkey = l_suppkey and ps_partkey = l_partkey and p_partkey = l_partkey and o_orderkey = l_orderkey and s_nationkey = n_nationkey and p_name like '%green%' ) as profit group by nation, o_year order by nation, o_year desc"
 "unsupported: cross-shard query with aggregates"
-Gen4 error: using aggregation on top of a *planbuilder.simpleProjection plan is not yet supported
+Gen4 error: aggregation on columns from different sources not supported yet
 
 # TPC-H query 10
 "select c_custkey, c_name, sum(l_extendedprice * (1 - l_discount)) as revenue, c_acctbal, n_name, c_address, c_phone, c_comment from customer, orders, lineitem, nation where c_custkey = o_custkey and l_orderkey = o_orderkey and o_orderdate >= date('1993-10-01') and o_orderdate < date('1993-10-01') + interval '3' month and l_returnflag = 'R' and c_nationkey = n_nationkey group by c_custkey, c_name, c_acctbal, c_phone, n_name, c_address, c_comment order by revenue desc limit 20"
@@ -657,7 +832,7 @@ Gen4 error: unsupported: in scatter query: complex aggregate expression
 
 # TPC-H query 13
 "select c_count, count(*) as custdist from ( select c_custkey, count(o_orderkey) from customer left outer join orders on c_custkey = o_custkey and o_comment not like '%special%requests%' group by c_custkey ) as c_orders(c_custkey, c_count) group by c_count order by custdist desc, c_count desc"
-"using aggregation on top of a *planbuilder.simpleProjection plan is not yet supported"
+"using aggregation on top of a *planbuilder.orderedAggregate plan is not yet supported"
 Gen4 plan same as above
 
 # TPC-H query 14
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index 50457d0ef1..ad8903d62a 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -510,7 +510,7 @@ Gen4 error: unsupported: in scatter query: complex order by expression: a + 1
 # aggregation on union
 "select sum(col) from (select col from user union all select col from unsharded) t"
 "unsupported: cross-shard query with aggregates"
-Gen4 error: using aggregation on top of a *planbuilder.simpleProjection plan is not yet supported
+Gen4 error: using aggregation on top of a *planbuilder.concatenateGen4 plan is not yet supported
 
 # systable union query in derived table with constraint on outside (without star projection)
 "select id from (select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'user_extra' union select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'music') `kcu` where `id` = 'primary'"
diff --git a/test.go b/test.go
index 7e851f1b6d..92c822be75 100755
--- a/test.go
+++ b/test.go
@@ -74,7 +74,7 @@ For example:
 // Flags
 var (
 	flavor           = flag.String("flavor", "mysql57", "comma-separated bootstrap flavor(s) to run against (when using Docker mode). Available flavors: all,"+flavors)
-	bootstrapVersion = flag.String("bootstrap-version", "6", "the version identifier to use for the docker images")
+	bootstrapVersion = flag.String("bootstrap-version", "8", "the version identifier to use for the docker images")
 	runCount         = flag.Int("runs", 1, "run each test this many times")
 	retryMax         = flag.Int("retry", 3, "max number of retries, to detect flaky tests")
 	logPass          = flag.Bool("log-pass", false, "log test output even if it passes")
diff --git a/test/templates/dockerfile.tpl b/test/templates/dockerfile.tpl
index dc595f2a81..d6dcb4fab2 100644
--- a/test/templates/dockerfile.tpl
+++ b/test/templates/dockerfile.tpl
@@ -1,4 +1,4 @@
-ARG bootstrap_version=6
+ARG bootstrap_version=8
 ARG image="vitess/bootstrap:${bootstrap_version}-{{.Platform}}"
 
 FROM "${image}"