Fix query list override issue on mysql restart

.github/workflows/cluster_endtoend_vttablet_prscomplex.yml

@@ -0,0 +1,118 @@
+# DO NOT MODIFY: THIS FILE IS GENERATED USING "make generate_ci_workflows"
+
+name: Cluster (vttablet_prscomplex)
+on: [push, pull_request]
+concurrency:
+  group: format('{0}-{1}', ${{ github.ref }}, 'Cluster (vttablet_prscomplex)')
+  cancel-in-progress: true
+
+env:
+  LAUNCHABLE_ORGANIZATION: "vitess"
+  LAUNCHABLE_WORKSPACE: "vitess-app"
+  GITHUB_PR_HEAD_SHA: "${{ github.event.pull_request.head.sha }}"
+
+jobs:
+  build:
+    name: Run endtoend tests on Cluster (vttablet_prscomplex)
+    runs-on: ubuntu-18.04
+
+    steps:
+    - name: Check if workflow needs to be skipped
+      id: skip-workflow
+      run: |
+        skip='false'
+        if [[ "${{github.event.pull_request}}" ==  "" ]] && [[ "${{github.ref}}" != "refs/heads/main" ]] && [[ ! "${{github.ref}}" =~ ^refs/heads/release-[0-9]+\.[0-9]$ ]] && [[ ! "${{github.ref}}" =~ "refs/tags/.*" ]]; then
+          skip='true'
+        fi
+        echo Skip ${skip}
+        echo "::set-output name=skip-workflow::${skip}"
+
+    - name: Check out code
+      if: steps.skip-workflow.outputs.skip-workflow == 'false'
+      uses: actions/checkout@v2
+
+    - name: Check for changes in relevant files
+      if: steps.skip-workflow.outputs.skip-workflow == 'false'
+      uses: frouioui/paths-filter@main
+      id: changes
+      with:
+        token: ''
+        filters: |
+          end_to_end:
+            - 'go/**/*.go'
+            - 'test.go'
+            - 'Makefile'
+            - 'build.env'
+            - 'go.[sumod]'
+            - 'proto/*.proto'
+            - 'tools/**'
+            - 'config/**'
+            - 'bootstrap.sh'
+            - '.github/workflows/cluster_endtoend_vttablet_prscomplex.yml'
+
+    - name: Set up Go
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/setup-go@v2
+      with:
+        go-version: 1.18.5
+
+    - name: Set up python
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      uses: actions/setup-python@v2
+
+    - name: Tune the OS
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        echo '1024 65535' | sudo tee -a /proc/sys/net/ipv4/ip_local_port_range
+        # Increase the asynchronous non-blocking I/O. More information at https://dev.mysql.com/doc/refman/5.7/en/innodb-parameters.html#sysvar_innodb_use_native_aio
+        echo "fs.aio-max-nr = 1048576" | sudo tee -a /etc/sysctl.conf
+        sudo sysctl -p /etc/sysctl.conf
+
+    - name: Get dependencies
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget eatmydata
+        sudo service mysql stop
+        sudo service etcd stop
+        sudo ln -s /etc/apparmor.d/usr.sbin.mysqld /etc/apparmor.d/disable/
+        sudo apparmor_parser -R /etc/apparmor.d/usr.sbin.mysqld
+        go mod download
+
+        # install JUnit report formatter
+        go install github.com/vitessio/go-junit-report@HEAD
+
+    - name: Setup launchable dependencies
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      run: |
+        # Get Launchable CLI installed. If you can, make it a part of the builder image to speed things up
+        pip3 install --user launchable~=1.0 > /dev/null
+
+        # verify that launchable setup is all correct.
+        launchable verify || true
+
+        # Tell Launchable about the build you are producing and testing
+        launchable record build --name "$GITHUB_RUN_ID" --source .
+
+    - name: Run cluster endtoend test
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
+      timeout-minutes: 30
+      run: |
+        # We set the VTDATAROOT to the /tmp folder to reduce the file path of mysql.sock file
+        # which musn't be more than 107 characters long.
+        export VTDATAROOT="/tmp/"
+        source build.env
+
+        set -x
+
+        # run the tests however you normally do, then produce a JUnit XML file
+        eatmydata -- go run test.go -docker=false -follow -shard vttablet_prscomplex | tee -a output.txt | go-junit-report -set-exit-code > report.xml
+
+    - name: Print test output and Record test result in launchable
+      if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true' && always()
+      run: |
+        # send recorded tests to launchable
+        launchable record tests --build "$GITHUB_RUN_ID" go-test . || true
+
+        # print test output
+        cat output.txt

go/test/endtoend/cluster/mysqlctl_process.go

@@ -65,7 +65,16 @@ func (mysqlctl *MysqlctlProcess) InitDb() (err error) {
 
 // Start executes mysqlctl command to start mysql instance
 func (mysqlctl *MysqlctlProcess) Start() (err error) {
-	tmpProcess, err := mysqlctl.StartProcess()
+	tmpProcess, err := mysqlctl.startProcess(true)
+	if err != nil {
+		return err
+	}
+	return tmpProcess.Wait()
+}
+
+// StartProvideInit executes mysqlctl command to start mysql instance
+func (mysqlctl *MysqlctlProcess) StartProvideInit(init bool) (err error) {
+	tmpProcess, err := mysqlctl.startProcess(init)
 	if err != nil {
 		return err
 	}
@@ -74,6 +83,10 @@ func (mysqlctl *MysqlctlProcess) Start() (err error) {
 
 // StartProcess starts the mysqlctl and returns the process reference
 func (mysqlctl *MysqlctlProcess) StartProcess() (*exec.Cmd, error) {
+	return mysqlctl.startProcess(true)
+}
+
+func (mysqlctl *MysqlctlProcess) startProcess(init bool) (*exec.Cmd, error) {
 	tmpProcess := exec.Command(
 		mysqlctl.Binary,
 		"--log_dir", mysqlctl.LogDirectory,
@@ -120,8 +133,10 @@ ssl_key={{.Dir}}/server-001-key.pem
 			tmpProcess.Env = append(tmpProcess.Env, "VTDATAROOT="+os.Getenv("VTDATAROOT"))
 		}
 
-		tmpProcess.Args = append(tmpProcess.Args, "init", "--",
-			"--init_db_sql_file", mysqlctl.InitDBFile)
+		if init {
+			tmpProcess.Args = append(tmpProcess.Args, "init", "--",
+				"--init_db_sql_file", mysqlctl.InitDBFile)
+		}
 	}
 	tmpProcess.Args = append(tmpProcess.Args, "start")
 	log.Infof("Starting mysqlctl with command: %v", tmpProcess.Args)

go/test/endtoend/reparent/prscomplex/main_test.go

@@ -0,0 +1,148 @@
+/*
+Copyright 2022 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package misc
+
+import (
+	"context"
+	_ "embed"
+	"flag"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/test/endtoend/cluster"
+	rutils "vitess.io/vitess/go/test/endtoend/reparent/utils"
+	"vitess.io/vitess/go/test/endtoend/utils"
+)
+
+var (
+	clusterInstance *cluster.LocalProcessCluster
+	vtParams        mysql.ConnParams
+	keyspaceName    = "ks"
+	cell            = "test"
+
+	//go:embed schema.sql
+	schemaSQL string
+)
+
+func TestMain(m *testing.M) {
+	defer cluster.PanicHandler(nil)
+	flag.Parse()
+
+	exitCode := func() int {
+		clusterInstance = cluster.NewCluster(cell, "localhost")
+		defer clusterInstance.Teardown()
+
+		// Start topo server
+		err := clusterInstance.StartTopo()
+		if err != nil {
+			return 1
+		}
+
+		// Start keyspace
+		keyspace := &cluster.Keyspace{
+			Name:      keyspaceName,
+			SchemaSQL: schemaSQL,
+		}
+		clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs,
+			"--queryserver-config-query-timeout=9000",
+			"--queryserver-config-pool-size=3",
+			"--queryserver-config-stream-pool-size=3",
+			"--queryserver-config-transaction-cap=2",
+			"--queryserver-config-transaction-timeout=20",
+			"--shutdown_grace_period=3",
+			"--queryserver-config-schema-change-signal=false")
+		err = clusterInstance.StartUnshardedKeyspace(*keyspace, 1, false)
+		if err != nil {
+			return 1
+		}
+
+		// Start vtgate
+		clusterInstance.VtGateExtraArgs = append(clusterInstance.VtGateExtraArgs,
+			"--planner-version=gen4",
+			"--mysql_default_workload=olap",
+			"--schema_change_signal=false")
+		err = clusterInstance.StartVtgate()
+		if err != nil {
+			return 1
+		}
+
+		vtParams = mysql.ConnParams{
+			Host: clusterInstance.Hostname,
+			Port: clusterInstance.VtgateMySQLPort,
+		}
+		return m.Run()
+	}()
+	os.Exit(exitCode)
+}
+
+/*
+TestAcquireSameConnID tests that a query started on a connection gets reconnected with a new connection.
+Another query acquires the old connection ID and does not override the query list maintained by the vttablet process.
+PRS should not fail as the query list is maintained appropriately.
+*/
+func TestAcquireSameConnID(t *testing.T) {
+	defer func() {
+		err := recover()
+		if err != nil {
+			require.Equal(t, "Fail in goroutine after TestAcquireSameConnID has completed", err)
+		}
+	}()
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	// start a reserved connection
+	utils.Exec(t, conn, "set sql_mode=''")
+	_ = utils.Exec(t, conn, "select connection_id()")
+
+	// restart the mysql to trigger reconnect on next query.
+	primTablet := clusterInstance.Keyspaces[0].Shards[0].PrimaryTablet()
+	err = primTablet.MysqlctlProcess.Stop()
+	require.NoError(t, err)
+	err = primTablet.MysqlctlProcess.StartProvideInit(false)
+	require.NoError(t, err)
+
+	go func() {
+		// this will trigger reconnect with a new connection id, which will be lower than the origin connection id.
+		_, _ = utils.ExecAllowError(t, conn, "select connection_id(), sleep(4000)")
+	}()
+	time.Sleep(5 * time.Second)
+
+	// run through 100 times to acquire new connection, this might override the original connection id.
+	var conn2 *mysql.Conn
+	for i := 0; i < 100; i++ {
+		conn2, err = mysql.Connect(ctx, &vtParams)
+		require.NoError(t, err)
+
+		utils.Exec(t, conn2, "set sql_mode=''")
+		// ReserveExecute
+		_ = utils.Exec(t, conn2, "select connection_id()")
+
+		// Execute
+		_ = utils.Exec(t, conn2, "select connection_id()")
+
+	}
+
+	// prs should happen without any error.
+	text, err := rutils.Prs(t, clusterInstance, clusterInstance.Keyspaces[0].Shards[0].Replica())
+	require.NoError(t, err, text)
+}

go/test/endtoend/reparent/prscomplex/schema.sql

@@ -0,0 +1,5 @@
+create table t1(
+                   id1 bigint,
+                   id2 bigint,
+                   primary key(id1)
+) Engine=InnoDB;