cherry pick pingcap#1390 to release-2.0

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>

_utils/terror_gen/errors_release.txt

@@ -127,6 +127,7 @@ ErrNoUUIDDirMatchGTID,[code=11120:class=functional:scope=internal:level=high], "
 ErrNoRelayPosMatchGTID,[code=11121:class=functional:scope=internal:level=high], "Message: no relay pos match gtid %s"
 ErrReaderReachEndOfFile,[code=11122:class=functional:scope=internal:level=low]
 ErrMetadataNoBinlogLoc,[code=11123:class=functional:scope=upstream:level=low], "Message: didn't found binlog location in dumped metadata file %s, Workaround: Please check log of dump unit, there maybe errors when read upstream binlog status"
+ErrPreviousGTIDNotExist,[code=11124:class=functional:scope=internal:level=high], "Message: no previous gtid event from binlog %s"
 ErrConfigCheckItemNotSupport,[code=20001:class=config:scope=internal:level=medium], "Message: checking item %s is not supported\n%s, Workaround: Please check `ignore-checking-items` config in task configuration file, which can be set including `all`/`dump_privilege`/`replication_privilege`/`version`/`binlog_enable`/`binlog_format`/`binlog_row_image`/`table_schema`/`schema_of_shard_tables`/`auto_increment_ID`."
 ErrConfigTomlTransform,[code=20002:class=config:scope=internal:level=medium], "Message: %s, Workaround: Please check the configuration file has correct TOML format."
 ErrConfigYamlTransform,[code=20003:class=config:scope=internal:level=medium], "Message: %s, Workaround: Please check the configuration file has correct YAML format."

chaos/cases/conf/source3.yaml

@@ -1,5 +1,5 @@
 source-id: "replica-03"
-enable-gtid: false
+enable-gtid: true
 enable-relay: true
 
 from:

errors.toml

@@ -772,6 +772,12 @@ description = ""
 workaround = "Please check log of dump unit, there maybe errors when read upstream binlog status"
 tags = ["upstream", "low"]
 
+[error.DM-functional-11124]
+message = "no previous gtid event from binlog %s"
+description = ""
+workaround = ""
+tags = ["internal", "high"]
+
 [error.DM-config-20001]
 message = "checking item %s is not supported\n%s"
 description = ""

pkg/binlog/event/event.go

@@ -851,3 +851,19 @@ func GenDummyEvent(header *replication.EventHeader, latestPos uint32, eventSize
  ev, err := GenQueryEvent(&headerClone, latestPos, 0, 0, 0, nil, nil, queryBytes)
  return ev, err
 }
+
+// GenHeartbeatEvent generates a heartbeat event.
+// ref: https://dev.mysql.com/doc/internals/en/heartbeat-event.html
+func GenHeartbeatEvent(header *replication.EventHeader) *replication.BinlogEvent {
+ // modify header
+ headerClone := *header // do a copy
+ headerClone.Flags = 0
+ headerClone.EventSize = 39
+ headerClone.Timestamp = 0
+ headerClone.EventType = replication.HEARTBEAT_EVENT
+
+ eventBytes := make([]byte, 39)
+ ev := &replication.BinlogEvent{Header: &headerClone, Event: &replication.GenericEvent{Data: eventBytes}}
+
+ return ev
+}

pkg/binlog/reader/util.go

@@ -140,3 +140,39 @@ func GetGTIDsForPos(ctx context.Context, r Reader, endPos gmysql.Position) (gtid
  }
  }
 }
+
+// GetPreviousGTIDFromGTIDSet tries to get previous GTID sets from Previous_GTID_EVENT GTID for the specified GITD Set.
+// events should be [fake_rotate_event,format_description_event,previous_gtids_event/mariadb_gtid_list_event]
+func GetPreviousGTIDFromGTIDSet(ctx context.Context, r Reader, gset gtid.Set) (gtid.Set, error) {
+ err := r.StartSyncByGTID(gset)
+ if err != nil {
+ return nil, err
+ }
+ defer r.Close()
+
+ for {
+ var e *replication.BinlogEvent
+ e, err = r.GetEvent(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ switch e.Header.EventType {
+ case replication.ROTATE_EVENT:
+ if e.Header.Timestamp == 0 || e.Header.LogPos == 0 { // fake rotate event
+ continue
+ }
+ return nil, terror.ErrPreviousGTIDNotExist.Generate(gset.String())
+ case replication.FORMAT_DESCRIPTION_EVENT:
+ continue
+ case replication.PREVIOUS_GTIDS_EVENT:
+ previousGset, err := event.GTIDsFromPreviousGTIDsEvent(e)
+ return previousGset, err
+ case replication.MARIADB_GTID_LIST_EVENT:
+ previousGset, err := event.GTIDsFromMariaDBGTIDListEvent(e)
+ return previousGset, err
+ default:
+ return nil, terror.ErrPreviousGTIDNotExist.Generate(gset.String())
+ }
+ }
+}

pkg/binlog/reader/util_test.go

@@ -62,3 +62,32 @@ func (t *testTCPReaderSuite) TestGetGTIDsForPos(c *C) {
  c.Assert(err, ErrorMatches, ".*invalid position .* or GTID not enabled in upstream.*")
  c.Assert(gs, IsNil)
 }
+
+// added to testTCPReaderSuite to re-use DB connection.
+func (t *testTCPReaderSuite) TestGetPreviousGTIDFromGTIDSet(c *C) {
+ var (
+ cfg = replication.BinlogSyncerConfig{
+ ServerID: serverIDs[1],
+ Flavor: flavor,
+ Host: t.host,
+ Port: uint16(t.port),
+ User: t.user,
+ Password: t.password,
+ UseDecimal: true,
+ VerifyChecksum: true,
+ }
+ ctx, cancel = context.WithTimeout(context.Background(), utils.DefaultDBTimeout)
+ )
+ defer cancel()
+
+ _, endGS, err := utils.GetMasterStatus(ctx, t.db, flavor)
+ c.Assert(err, IsNil)
+
+ r1 := NewTCPReader(cfg)
+ c.Assert(r1, NotNil)
+ defer r1.Close()
+
+ gs, err := GetPreviousGTIDFromGTIDSet(ctx, r1, endGS)
+ c.Assert(err, IsNil)
+ c.Assert(endGS.Contain(gs), IsTrue)
+}