Reject TwoPC calls if semi-sync is not enabled

go/test/endtoend/transaction/twopc/twopc_test.go

@@ -1030,6 +1030,24 @@ func TestReadingUnresolvedTransactions(t *testing.T) {
 	}
 }
 
+// TestSemiSyncRequiredWithTwoPC tests that semi-sync is required when using two-phase commit.
+func TestSemiSyncRequiredWithTwoPC(t *testing.T) {
+	out, err := clusterInstance.VtctldClientProcess.ExecuteCommandWithOutput("SetKeyspaceDurabilityPolicy", keyspaceName, "--durability-policy=none")
+	require.NoError(t, err, out)
+	defer clusterInstance.VtctldClientProcess.ExecuteCommandWithOutput("SetKeyspaceDurabilityPolicy", keyspaceName, "--durability-policy=semi_sync")
+
+	// After changing the durability policy for the given keyspace to none, we try to PRS.
+	// This call should fail.
+	shard := clusterInstance.Keyspaces[0].Shards[2]
+	newPrimary := shard.Vttablets[1]
+	output, err := clusterInstance.VtctldClientProcess.ExecuteCommandWithOutput(
+		"PlannedReparentShard",
+		fmt.Sprintf("%s/%s", keyspaceName, shard.Name),
+		"--new-primary", newPrimary.Alias)
+	require.Error(t, err)
+	require.Contains(t, output, "two-pc is enabled, but semi-sync is not")
+}
+
 // TestDisruptions tests that atomic transactions persevere through various disruptions.
 func TestDisruptions(t *testing.T) {
 	testcases := []struct {

go/vt/vterrors/code.go

@@ -97,6 +97,7 @@ var (
 	VT09023 = errorWithoutState("VT09023", vtrpcpb.Code_FAILED_PRECONDITION, "could not map %v to a keyspace id", "Unable to determine the shard for the given row.")
 	VT09024 = errorWithoutState("VT09024", vtrpcpb.Code_FAILED_PRECONDITION, "could not map %v to a unique keyspace id: %v", "Unable to determine the shard for the given row.")
 	VT09025 = errorWithoutState("VT09025", vtrpcpb.Code_FAILED_PRECONDITION, "atomic transaction error: %v", "Error in atomic transactions")
+	VT09026 = errorWithoutState("VT09026", vtrpcpb.Code_FAILED_PRECONDITION, "two-pc is enabled, but semi-sync is not", "Two-PC requires semi-sync but it is not enabled")
 
 	VT10001 = errorWithoutState("VT10001", vtrpcpb.Code_ABORTED, "foreign key constraints are not allowed", "Foreign key constraints are not allowed, see https://vitess.io/blog/2021-06-15-online-ddl-why-no-fk/.")
 	VT10002 = errorWithoutState("VT10002", vtrpcpb.Code_ABORTED, "atomic distributed transaction not allowed: %s", "The distributed transaction cannot be committed. A rollback decision is taken.")

go/vt/vttablet/tabletmanager/rpc_replication.go

@@ -22,11 +22,10 @@ import (
 	"strings"
 	"time"
 
+	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/mysql/replication"
 	"vitess.io/vitess/go/mysql/sqlerror"
 	"vitess.io/vitess/go/protoutil"
-
-	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/mysqlctl"
 	"vitess.io/vitess/go/vt/proto/vtrpc"
@@ -348,6 +347,17 @@ func (tm *TabletManager) InitPrimary(ctx context.Context, semiSync bool) (string
 	}
 	defer tm.unlock()
 
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(ctx, semiSync)
+	if err != nil {
+		return "", err
+	}
+
+	// Check if two-pc is enabled but semi-sync is not.
+	// If so, we return an error, because atomic transactions require semi-sync for correctness.
+	if tm.QueryServiceControl.TwoPCEnabled() && semiSyncAction != SemiSyncActionSet {
+		return "", vterrors.VT09026()
+	}
+
 	// Setting super_read_only `OFF` so that we can run the DDL commands
 	if _, err := tm.MysqlDaemon.SetSuperReadOnly(ctx, false); err != nil {
 		if sqlErr, ok := err.(*sqlerror.SQLError); ok && sqlErr.Number() == sqlerror.ERUnknownSystemVariable {
@@ -369,11 +379,6 @@ func (tm *TabletManager) InitPrimary(ctx context.Context, semiSync bool) (string
 		return "", err
 	}
 
-	semiSyncAction, err := tm.convertBoolToSemiSyncAction(ctx, semiSync)
-	if err != nil {
-		return "", err
-	}
-
 	// Set the server read-write, from now on we can accept real
 	// client writes. Note that if semi-sync replication is enabled,
 	// we'll still need some replicas to be able to commit transactions.
@@ -910,12 +915,18 @@ func (tm *TabletManager) PromoteReplica(ctx context.Context, semiSync bool) (str
 	}
 	defer tm.unlock()
 
-	pos, err := tm.MysqlDaemon.Promote(ctx, tm.hookExtraEnv())
+	semiSyncAction, err := tm.convertBoolToSemiSyncAction(ctx, semiSync)
 	if err != nil {
 		return "", err
 	}
 
-	semiSyncAction, err := tm.convertBoolToSemiSyncAction(ctx, semiSync)
+	// Check if two-pc is enabled but semi-sync is not.
+	// If so, we return an error, because atomic transactions require semi-sync for correctness.
+	if tm.QueryServiceControl.TwoPCEnabled() && semiSyncAction != SemiSyncActionSet {
+		return "", vterrors.VT09026()
+	}
+
+	pos, err := tm.MysqlDaemon.Promote(ctx, tm.hookExtraEnv())
 	if err != nil {
 		return "", err
 	}

go/vt/vttablet/tabletserver/controller.go

@@ -93,6 +93,9 @@ type Controller interface {
 	// CheckThrottler
 	CheckThrottler(ctx context.Context, appName string, flags *throttle.CheckFlags) *throttle.CheckResult
 	GetThrottlerStatus(ctx context.Context) *throttle.ThrottlerStatus
+
+	// TwoPCEnabled returns if two pc is enabled or not.
+	TwoPCEnabled() bool
 }
 
 // Ensure TabletServer satisfies Controller interface.

go/vt/vttablet/tabletserver/tabletserver.go

@@ -1692,6 +1692,11 @@ func (tsv *TabletServer) GetThrottlerStatus(ctx context.Context) *throttle.Throt
 	return r
 }
 
+// TwoPCEnabled returns whether TwoPC is enabled or not.
+func (tsv *TabletServer) TwoPCEnabled() bool {
+	return tsv.config.TwoPCEnable
+}
+
 // HandlePanic is part of the queryservice.QueryService interface
 func (tsv *TabletServer) HandlePanic(err *error) {
 	if x := recover(); x != nil {

go/vt/vttablet/tabletservermock/controller.go

@@ -226,6 +226,11 @@ func (tqsc *Controller) GetThrottlerStatus(ctx context.Context) *throttle.Thrott
 	return nil
 }
 
+// TwoPCEnabled returns whether TwoPC is enabled or not.
+func (tqsc *Controller) TwoPCEnabled() bool {
+	return false
+}
+
 // EnterLameduck implements tabletserver.Controller.
 func (tqsc *Controller) EnterLameduck() {
 	tqsc.mu.Lock()