[release-17.0] CI: fix onlineddl_scheduler flakiness (#13754) (#13760)

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Co-authored-by: vitess-bot[bot] <108069721+vitess-bot[bot]@users.noreply.github.com>

go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go

@@ -167,6 +167,38 @@ func TestParseTableName(t *testing.T) {
  }
 }
 
+func waitForReadyToComplete(t *testing.T, uuid string, expected bool) {
+ ctx, cancel := context.WithTimeout(context.Background(), normalWaitTime)
+ defer cancel()
+
+ ticker := time.NewTicker(time.Second)
+ defer ticker.Stop()
+ for {
+
+ rs := onlineddl.ReadMigrations(t, &vtParams, uuid)
+ require.NotNil(t, rs)
+ for _, row := range rs.Named().Rows {
+ readyToComplete := row.AsInt64("ready_to_complete", 0)
+ if expected == (readyToComplete > 0) {
+ // all good. This is what we waited for
+ if expected {
+ // if migration is ready to complete, the nthe timestamp should be non-null
+ assert.False(t, row["ready_to_complete_timestamp"].IsNull())
+ } else {
+ assert.True(t, row["ready_to_complete_timestamp"].IsNull())
+ }
+
+ return
+ }
+ }
+ select {
+ case <-ticker.C:
+ case <-ctx.Done():
+ }
+ require.NoError(t, ctx.Err())
+ }
+}
+
 func TestMain(m *testing.M) {
  defer cluster.PanicHandler(nil)
  flag.Parse()
@@ -557,13 +589,7 @@ func testScheduler(t *testing.T) {
  })
  t.Run("check ready to complete (before)", func(t *testing.T) {
  for _, uuid := range []string{t1uuid, t2uuid} {
- rs := onlineddl.ReadMigrations(t, &vtParams, uuid)
- require.NotNil(t, rs)
- for _, row := range rs.Named().Rows {
- readyToComplete := row.AsInt64("ready_to_complete", 0)
- assert.Equal(t, int64(0), readyToComplete)
- assert.True(t, row["ready_to_complete_timestamp"].IsNull())
- }
+ waitForReadyToComplete(t, uuid, false)
  }
  })
  t.Run("unthrottle, expect t2 running", func(t *testing.T) {
@@ -595,34 +621,23 @@ func testScheduler(t *testing.T) {
  })
  t.Run("check ready to complete (after)", func(t *testing.T) {
  for _, uuid := range []string{t1uuid, t2uuid} {
- rs := onlineddl.ReadMigrations(t, &vtParams, uuid)
- require.NotNil(t, rs)
- for _, row := range rs.Named().Rows {
- readyToComplete := row.AsInt64("ready_to_complete", 0)
- assert.Equal(t, int64(1), readyToComplete)
- assert.False(t, row["ready_to_complete_timestamp"].IsNull())
- }
+ waitForReadyToComplete(t, uuid, true)
  }
  })
 
  testTableCompletionTimes(t, t2uuid, t1uuid)
  })
  t.Run("REVERT both tables concurrent, postponed", func(t *testing.T) {
- t1uuid = testRevertMigration(t, createRevertParams(t1uuid, ddlStrategy+" -allow-concurrent -postpone-completion", "vtgate", "", true))
- t2uuid = testRevertMigration(t, createRevertParams(t2uuid, ddlStrategy+" -allow-concurrent -postpone-completion", "vtgate", "", true))
+ t1uuid = testRevertMigration(t, createRevertParams(t1uuid, ddlStrategy+" --allow-concurrent --postpone-completion", "vtgate", "", true))
+ t2uuid = testRevertMigration(t, createRevertParams(t2uuid, ddlStrategy+" --allow-concurrent --postpone-completion", "vtgate", "", true))
 
  testAllowConcurrent(t, "t1", t1uuid, 1)
  t.Run("expect both migrations to run", func(t *testing.T) {
  onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t1uuid, normalWaitTime, schema.OnlineDDLStatusRunning)
  onlineddl.WaitForMigrationStatus(t, &vtParams, shards, t2uuid, normalWaitTime, schema.OnlineDDLStatusRunning)
  })
  t.Run("test ready-to-complete", func(t *testing.T) {
- rs := onlineddl.ReadMigrations(t, &vtParams, t1uuid)
- require.NotNil(t, rs)
- for _, row := range rs.Named().Rows {
- readyToComplete := row.AsInt64("ready_to_complete", 0)
- assert.Equal(t, int64(1), readyToComplete)
- }
+ waitForReadyToComplete(t, t1uuid, true)
  })
  t.Run("complete t2", func(t *testing.T) {
  // now that both are running, let's unblock t2. We expect it to complete.
@@ -760,12 +775,7 @@ func testScheduler(t *testing.T) {
  onlineddl.CheckMigrationStatus(t, &vtParams, shards, drop1uuid, schema.OnlineDDLStatusReady)
  })
  t.Run("t3 ready to complete", func(t *testing.T) {
- rs := onlineddl.ReadMigrations(t, &vtParams, drop1uuid)
- require.NotNil(t, rs)
- for _, row := range rs.Named().Rows {
- readyToComplete := row.AsInt64("ready_to_complete", 0)
- assert.Equal(t, int64(1), readyToComplete)
- }
+ waitForReadyToComplete(t, drop1uuid, true)
  })
  t.Run("t3drop complete", func(t *testing.T) {
  // drop3 migration should not block. It can run concurrently to t1, and does not conflict